path: root/hicn-light/src/hicn/io/streamConnection.c

/*
 * Copyright (c) 2017-2019 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Common activity for STREAM based listeners.
 */

#include <errno.h>
#include <hicn/hicn-light/config.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>

#include <parc/algol/parc_Hash.h>
#include <hicn/core/connection.h>
#include <hicn/core/forwarder.h>
#include <hicn/core/message.h>
#include <hicn/io/streamConnection.h>

#include <parc/algol/parc_Memory.h>
#include <parc/assert/parc_Assert.h>
#include <hicn/core/messageHandler.h>

#include <hicn/utils/commands.h>

#include <hicn/hicn.h>
// 128 KB output queue
#define OUTPUT_QUEUE_BYTES (128 * 1024)

static void _conn_readcb(PARCEventQueue *bufferEventVector, PARCEventType type,
                         void *ioOpsVoid);

static void _conn_eventcb(PARCEventQueue *bufferEventVector,
                          PARCEventQueueEventType events, void *ioOpsVoid);

typedef struct stream_state {
  Forwarder *forwarder;
  char * interfaceName;
  Logger *logger;

  int fd;

  AddressPair *addressPair;
  PARCEventQueue *bufferEventVector;

  bool isLocal;
  bool isUp;
  bool isClosed;
  unsigned id;

  size_t nextMessageLength;

  /* This information would better be stored in the connection data structure
   * but it is currently not reachable from within the implementation. */
  connection_state_t state;
  connection_state_t admin_state;
#ifdef WITH_POLICY
  uint32_t priority;
#endif /* WITH_POLICY */
} _StreamState;

// Prototypes
static bool _streamConnection_Send(IoOperations *ops, const Address *nexthop,
    Message *message);
static bool _streamConnection_SendIOVBuffer(IoOperations *ops, struct
    iovec *msg, size_t size);
static const Address *_streamConnection_GetRemoteAddress(
    const IoOperations *ops);
static const AddressPair *_streamConnection_GetAddressPair(
    const IoOperations *ops);
static unsigned _streamConnection_GetConnectionId(const IoOperations *ops);
static bool _streamConnection_IsUp(const IoOperations *ops);
static bool _streamConnection_IsLocal(const IoOperations *ops);
static void _streamConnection_DestroyOperations(IoOperations **opsPtr);

static void _setConnectionState(_StreamState *stream, bool isUp);
static list_connections_type _streamConnection_GetConnectionType(
    const IoOperations *ops);
static Ticks _sendProbe(IoOperations *ops, unsigned probeType,
                        uint8_t *message);
static connection_state_t _streamConnection_getState(const IoOperations *ops);
static void _streamConnection_setState(IoOperations *ops, connection_state_t state);
static connection_state_t _streamConnection_getAdminState(const IoOperations *ops);
static void _streamConnection_setAdminState(IoOperations *ops, connection_state_t admin_state);
#ifdef WITH_POLICY
static uint32_t _streamConnection_getPriority(const IoOperations *ops);
static void _streamConnection_setPriority(IoOperations *ops, uint32_t priority);
#endif /* WITH_POLICY */
static const char * _streamConnection_getInterfaceName(const IoOperations *ops);

/*
 * This assigns a unique pointer to the void * which we use
 * as a GUID for this class.
 */
static const void *_ioOperationsGuid = __FILE__;

/*
 * Return our GUID
 */
static const void *_streamConnection_Class(const IoOperations *ops) {
  return _ioOperationsGuid;
}

static IoOperations _template = {
    .closure = NULL,
    .send = &_streamConnection_Send,
    .sendIOVBuffer = &_streamConnection_SendIOVBuffer,
    .getRemoteAddress = &_streamConnection_GetRemoteAddress,
    .getAddressPair = &_streamConnection_GetAddressPair,
    .getConnectionId = &_streamConnection_GetConnectionId,
    .isUp = &_streamConnection_IsUp,
    .isLocal = &_streamConnection_IsLocal,
    .destroy = &_streamConnection_DestroyOperations,
    .class = &_streamConnection_Class,
    .getConnectionType = &_streamConnection_GetConnectionType,
    .sendProbe = &_sendProbe,
    .getState = &_streamConnection_getState,
    .setState = &_streamConnection_setState,
    .getAdminState = &_streamConnection_getAdminState,
    .setAdminState = &_streamConnection_setAdminState,
#ifdef WITH_POLICY
    .getPriority = &_streamConnection_getPriority,
    .setPriority = &_streamConnection_setPriority,
#endif /* WITH_POLICY */
    .getInterfaceName = &_streamConnection_getInterfaceName,
};

IoOperations *streamConnection_AcceptConnection(Forwarder *forwarder, int fd,
                                                AddressPair *pair,
                                                bool isLocal) {
  _StreamState *stream = parcMemory_AllocateAndClear(sizeof(_StreamState));
  parcAssertNotNull(stream, "parcMemory_AllocateAndClear(%zu) returned NULL",
                    sizeof(_StreamState));

  Dispatcher *dispatcher = forwarder_GetDispatcher(forwarder);
  PARCEventScheduler *eventBase = dispatcher_GetEventScheduler(dispatcher);
  stream->bufferEventVector = parcEventQueue_Create(
      eventBase, fd,
      PARCEventQueueOption_CloseOnFree | PARCEventQueueOption_DeferCallbacks);

  stream->forwarder = forwarder;
  stream->interfaceName = NULL;
  stream->logger = logger_Acquire(forwarder_GetLogger(forwarder));
  stream->fd = fd;
  stream->id = forwarder_GetNextConnectionId(forwarder);
  stream->addressPair = pair;
  stream->isClosed = false;

#ifdef WITH_POLICY
  stream->priority = 0;
#endif /* WITH_POLICY */

  // allocate a connection
  IoOperations *io_ops = parcMemory_AllocateAndClear(sizeof(IoOperations));
  parcAssertNotNull(io_ops, "parcMemory_AllocateAndClear(%zu) returned NULL",
                    sizeof(IoOperations));
  memcpy(io_ops, &_template, sizeof(IoOperations));
  io_ops->closure = stream;
  stream->isLocal = isLocal;

  parcEventQueue_SetCallbacks(stream->bufferEventVector, _conn_readcb, NULL,
                              _conn_eventcb, (void *)io_ops);
  parcEventQueue_Enable(stream->bufferEventVector, PARCEventType_Read);

  messenger_Send(forwarder_GetMessenger(stream->forwarder),
                 missive_Create(MissiveType_ConnectionCreate, stream->id));

  // As we are acceting a connection, we begin in the UP state
  _setConnectionState(stream, true);

  if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                        PARCLogLevel_Debug)) {
    char *pair_str = addressPair_ToString(pair);
    logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Debug, __func__,
               "StreamConnection %p accept for address pair %s", (void *)stream,
               pair_str);
    free(pair_str);
  }

  return io_ops;
}

IoOperations *streamConnection_OpenConnection(Forwarder *forwarder,
                                              AddressPair *pair, bool isLocal) {
  parcAssertNotNull(forwarder, "Parameter hicn-light must be non-null");
  parcAssertNotNull(pair, "Parameter pair must be non-null");

  // if there's an error on the bind or connect, will return NULL
  PARCEventQueue *bufferEventVector =
      dispatcher_StreamBufferConnect(forwarder_GetDispatcher(forwarder), pair);
  if (bufferEventVector == NULL) {
    // error opening connection
    return NULL;
  }

  _StreamState *stream = parcMemory_AllocateAndClear(sizeof(_StreamState));
  parcAssertNotNull(stream, "parcMemory_AllocateAndClear(%zu) returned NULL",
                    sizeof(_StreamState));

  stream->forwarder = forwarder;
  stream->interfaceName = NULL;
  stream->logger = logger_Acquire(forwarder_GetLogger(forwarder));
  stream->fd = parcEventQueue_GetFileDescriptor(bufferEventVector);
  stream->bufferEventVector = bufferEventVector;
  stream->id = forwarder_GetNextConnectionId(forwarder);
  stream->addressPair = pair;
  stream->isClosed = false;

  // allocate a connection
  IoOperations *io_ops = parcMemory_AllocateAndClear(sizeof(IoOperations));
  parcAssertNotNull(io_ops, "parcMemory_AllocateAndClear(%zu) returned NULL",
                    sizeof(IoOperations));
  memcpy(io_ops, &_template, sizeof(IoOperations));
  io_ops->closure = stream;
  stream->isLocal = isLocal;

  parcEventQueue_SetCallbacks(stream->bufferEventVector, _conn_readcb, NULL,
                              _conn_eventcb, (void *)io_ops);
  parcEventQueue_Enable(stream->bufferEventVector, PARCEventType_Read);

  // we start in DOWN state, until remote side answers
  messenger_Send(forwarder_GetMessenger(stream->forwarder),
                 missive_Create(MissiveType_ConnectionCreate, stream->id));
  _setConnectionState(stream, false);

  if (logger_IsLoggable(stream->logger, LoggerFacility_IO, PARCLogLevel_Info)) {
    char *pair_str = addressPair_ToString(pair);
    logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Info, __func__,
               "StreamConnection %p connect for address pair %s",
               (void *)stream, pair_str);
    free(pair_str);
  }

  return io_ops;
}

static void _streamConnection_DestroyOperations(IoOperations **opsPtr) {
  parcAssertNotNull(opsPtr, "Parameter opsPtr must be non-null double pointer");
  parcAssertNotNull(*opsPtr,
                    "Parameter opsPtr must dereference to non-null pointer");

  IoOperations *ops = *opsPtr;
  parcAssertNotNull(ioOperations_GetClosure(ops),
                    "ops->context must not be null");

  _StreamState *stream = (_StreamState *)ioOperations_GetClosure(ops);

  parcEventQueue_Destroy(&stream->bufferEventVector);

  addressPair_Release(&stream->addressPair);

  if (!stream->isClosed) {
    stream->isClosed = true;
    messenger_Send(forwarder_GetMessenger(stream->forwarder),
                   missive_Create(MissiveType_ConnectionClosed, stream->id));
  }

  messenger_Send(forwarder_GetMessenger(stream->forwarder),
                 missive_Create(MissiveType_ConnectionDestroyed, stream->id));

  if (logger_IsLoggable(stream->logger, LoggerFacility_IO, PARCLogLevel_Info)) {
    logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Info, __func__,
               "StreamConnection %p destroyed", (void *)stream);
  }

  logger_Release(&stream->logger);
  parcMemory_Deallocate((void **)&stream);
  parcMemory_Deallocate((void **)&ops);

  *opsPtr = NULL;
}

static bool _streamConnection_IsUp(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->isUp;
}

static bool _streamConnection_IsLocal(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->isLocal;
}

static const Address *_streamConnection_GetRemoteAddress(
    const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return addressPair_GetRemote(stream->addressPair);
}

static const AddressPair *_streamConnection_GetAddressPair(
    const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->addressPair;
}

static unsigned _streamConnection_GetConnectionId(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->id;
}

bool _streamConnection_SendIOVBuffer(IoOperations *ops,
    struct iovec * message, size_t size) {
  parcAssertNotNull(ops, "Parameter ops must be non-null");
  parcAssertNotNull(message, "Parameter message must be non-null");

  _StreamState *conn = (_StreamState *)ioOperations_GetClosure(ops);

  if (!conn->isUp) {
    if (logger_IsLoggable(conn->logger, LoggerFacility_IO,
                          PARCLogLevel_Error)) {
      logger_Log(
          conn->logger, LoggerFacility_IO, PARCLogLevel_Error, __func__,
          "connid %u tried to send to down connection (isUp %d isClosed %d)",
          conn->id, conn->isUp, conn->isClosed);
    }
    return false;
  }

  PARCEventBuffer *buffer =
      parcEventBuffer_GetQueueBufferOutput(conn->bufferEventVector);
  size_t buffer_backlog = parcEventBuffer_GetLength(buffer);
  parcEventBuffer_Destroy(&buffer);

  if (buffer_backlog >= OUTPUT_QUEUE_BYTES) {
    if (logger_IsLoggable(conn->logger, LoggerFacility_IO,
                          PARCLogLevel_Warning)) {
      logger_Log(conn->logger, LoggerFacility_IO, PARCLogLevel_Warning,
                 __func__,
                 "connid %u Writing to buffer backlog %zu bytes DROP MESSAGE",
                 conn->id, buffer_backlog);
    }
    return false;
  }

  if (logger_IsLoggable(conn->logger, LoggerFacility_IO,
                        PARCLogLevel_Debug)) {
    size_t length = 0;
    for (int i = 0; i < size; i++)
      length += message[i].iov_len;

    logger_Log( conn->logger, LoggerFacility_IO, PARCLogLevel_Debug, __func__,
            "connid %u Writing %zu bytes to buffer with backlog %zu bytes",
            conn->id, length, buffer_backlog);
  }

  /* Write directly into the parcEventQueue without passing through message */
  for (int i = 0; i < size; i++) {
    if (parcEventQueue_Write(conn->bufferEventVector, message[i].iov_base,
                message[i].iov_len) != 0)
        return false;
  }

  return true;
}

/**
 * @function streamConnection_Send
 * @abstract Non-destructive send of the message.
 * @discussion
 *   Send uses message_CopyToStreamBuffer, which is a non-destructive write.
 *   The send may fail if there's no buffer space in the output queue.
 *
 * @param dummy is ignored.  A stream has only one peer.
 * @return <#return#>
 */
static bool _streamConnection_Send(IoOperations *ops, const Address *dummy,
                                   Message *message) {
  parcAssertNotNull(ops, "Parameter ops must be non-null");
  parcAssertNotNull(message, "Parameter message must be non-null");
  _StreamState *stream = (_StreamState *)ioOperations_GetClosure(ops);

  bool success = false;
  if (stream->isUp) {
    PARCEventBuffer *buffer =
        parcEventBuffer_GetQueueBufferOutput(stream->bufferEventVector);
    size_t buffer_backlog = parcEventBuffer_GetLength(buffer);
    parcEventBuffer_Destroy(&buffer);

    if (buffer_backlog < OUTPUT_QUEUE_BYTES) {
      if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                            PARCLogLevel_Debug)) {
        logger_Log(
            stream->logger, LoggerFacility_IO, PARCLogLevel_Debug, __func__,
            "connid %u Writing %zu bytes to buffer with backlog %zu bytes",
            stream->id, message_Length(message), buffer_backlog);
      }

      int failure = message_Write(stream->bufferEventVector, message);
      if (failure == 0) {
        success = true;
      }
    } else {
      if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                            PARCLogLevel_Warning)) {
        logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Warning,
                   __func__,
                   "connid %u Writing to buffer backlog %zu bytes DROP MESSAGE",
                   stream->id, buffer_backlog);
      }
    }
  } else {
    if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                          PARCLogLevel_Error)) {
      logger_Log(
          stream->logger, LoggerFacility_IO, PARCLogLevel_Error, __func__,
          "connid %u tried to send to down connection (isUp %d isClosed %d)",
          stream->id, stream->isUp, stream->isClosed);
    }
  }

  return success;
}

list_connections_type _streamConnection_GetConnectionType(
    const IoOperations *ops) {
  return CONN_TCP;
}

static Ticks _sendProbe(IoOperations *ops, unsigned probeType,
                        uint8_t *message) {
  // we don't need to implemet this here, it is a local connection
  return 0;
}

// =================================================================
// the actual I/O functions

int _isACommand(PARCEventBuffer *input) {
  size_t bytesAvailable = parcEventBuffer_GetLength(input);
  parcAssertTrue(bytesAvailable >= sizeof(header_control_message),
                 "Called with too short an input: %zu", bytesAvailable);

  uint8_t *msg = parcEventBuffer_Pullup(input, bytesAvailable);
  // read first byte of the header

  // first byte: must be a REQUEST_LIGHT
  if (msg[0] != REQUEST_LIGHT) {
    return LAST_COMMAND_VALUE;
  }

  // second byte: must be a command_id
  if (msg[1] < 0 || msg[1] >= LAST_COMMAND_VALUE) {
    return LAST_COMMAND_VALUE;
  }

  return msg[1];
}

PARCEventBuffer *_tryReadControlMessage(_StreamState *stream,
                                        PARCEventBuffer *input,
                                        command_id command,
                                        struct iovec **request) {
  size_t bytesAvailable = parcEventBuffer_GetLength(input);

  if (stream->nextMessageLength == 0) {
    stream->nextMessageLength =
        sizeof(header_control_message) +
        payloadLengthDaemon(command);  // consider the whole packet.
  }

  if (bytesAvailable >= stream->nextMessageLength) {
    PARCEventBuffer *message = parcEventBuffer_Create();
    int bytesRead = parcEventBuffer_ReadIntoBuffer(input, message,
                                                   stream->nextMessageLength);
    parcAssertTrue(bytesRead == stream->nextMessageLength,
                   "Partial read, expected %zu got %d",
                   stream->nextMessageLength, bytesRead);

    uint8_t *control =
        parcEventBuffer_Pullup(message, stream->nextMessageLength);
    if (!(*request = (struct iovec *)parcMemory_AllocateAndClear(
              sizeof(struct iovec) * 2))) {
      return NULL;
    }
    (*request)[0].iov_base = control;  // header
    (*request)[0].iov_len = sizeof(header_control_message);
    if (payloadLengthDaemon(command) > 0) {
      (*request)[1].iov_base =
          control + sizeof(header_control_message);  // payload
    } else {
      (*request)[1].iov_base = NULL;
    }
    (*request)[1].iov_len = payloadLengthDaemon(command);
    // now reset message length for next packet

    stream->nextMessageLength = 0;

    return message;
  }

  return NULL;
}

static bool _isAnHicnPacket(PARCEventBuffer *input) {
  size_t bytesAvailable = parcEventBuffer_GetLength(input);
  parcAssertTrue(bytesAvailable >= sizeof(header_control_message),
                 "Called with too short an input: %zu", bytesAvailable);

  uint8_t *fh = parcEventBuffer_Pullup(input, sizeof(header_control_message));
  return messageHandler_IsValidHicnPacket(fh);
}

static Message *_readMessage(_StreamState *stream, Ticks time,
                             PARCEventBuffer *input) {
  Message *message = message_CreateFromEventBuffer(
      input, stream->nextMessageLength, stream->id, time, stream->logger);

  return message;
}

static void _startNewMessage(_StreamState *stream, PARCEventBuffer *input,
                             size_t inputBytesAvailable) {
  parcAssertTrue(stream->nextMessageLength == 0,
                 "Invalid state, nextMessageLength not zero: %zu",
                 stream->nextMessageLength);
  parcAssertTrue(inputBytesAvailable >= sizeof(header_control_message),
                 "read_length not a whole fixed header!: %zd",
                 inputBytesAvailable);

  // this linearizes the first messageHandler_GetIPv6HeaderLength() bytes of the
  // input buffer's iovecs and returns a pointer to it.
  uint8_t *fh = parcEventBuffer_Pullup(input, sizeof(header_control_message));

  // Calculate the total message size based on the fixed header
  stream->nextMessageLength = messageHandler_GetTotalPacketLength(fh);
}

static Message *_tryReadMessage(PARCEventBuffer *input, _StreamState *stream) {
  size_t bytesAvailable = parcEventBuffer_GetLength(input);
  parcAssertTrue(bytesAvailable >= sizeof(header_control_message),
                 "Called with too short an input: %zu", bytesAvailable);

  if (stream->nextMessageLength == 0) {
    _startNewMessage(stream, input, bytesAvailable);
  }

  // This is not an ELSE statement.  We can both start a new message then
  // check if there's enough bytes to read the whole thing.

  if (bytesAvailable >= stream->nextMessageLength) {
    Message *message =
        _readMessage(stream, forwarder_GetTicks(stream->forwarder), input);

    // now reset message length for next packet
    stream->nextMessageLength = 0;

    return message;
  }

  return NULL;
}

/**
 * @function conn_readcb
 * @abstract Event callback for reads
 * @discussion
 *   Will read messages off the input.  Continues reading as long as we
 *   can get a header to determine the next message length or as long as we
 *   can read a complete message.
 *
 *   This function manipulates the read low water mark.  (1) read a fixed header
 * plus complete message, then set the low water mark to FIXED_HEADER_LEN.  (2)
 * read a fixed header, but not a complete message, then set low water mark to
 * the total mesage length.  Using the low water mark like this means the buffer
 * event will only trigger on meaningful byte boundaries when we can get actual
 *   work done.
 *
 * @param <#param1#>
 * @return <#return#>
 */
static void _conn_readcb(PARCEventQueue *event, PARCEventType type,
                         void *ioOpsVoid) {
  command_id command;
  IoOperations *ops = (IoOperations *)ioOpsVoid;
  _StreamState *stream = (_StreamState *)ioOperations_GetClosure(ops);

  PARCEventBuffer *input = parcEventBuffer_GetQueueBufferInput(event);

  // drain the input buffer

  // notice that we always try to read at least 8 bytes
  // (sizeof(header_control_message)). This is enough to read the length of all
  // kind of packets
  while (parcEventBuffer_GetLength(input) >= sizeof(header_control_message) &&
         parcEventBuffer_GetLength(input) >= stream->nextMessageLength) {

    if ((command = _isACommand(input)) != LAST_COMMAND_VALUE) {
      struct iovec *rx;
      // Get message from the stream and set the stream->nextMessageLength
      PARCEventBuffer *message =
          _tryReadControlMessage(stream, input, command, &rx);
      // If received correctly the whole message, send to dispatcher
      if (message) {
        forwarder_ReceiveCommand(stream->forwarder, command, rx, stream->id);
        parcMemory_Deallocate((void **)&rx);
        parcEventBuffer_Destroy(&message);
      }

    } else if (_isAnHicnPacket(input)) {
      // this is an Hicn packet (here we should distinguish between IPv4 and
      // IPv6 tryReadMessage may set nextMessageLength
      Message *message = _tryReadMessage(input, stream);

      if (message) {
        forwarder_Receive(stream->forwarder, message);
      }

    } else {
      parcAssertTrue(false,
                     "(Local stream connection) malformed packet received");
    }
  }

  if (stream->nextMessageLength == 0) {
    // we don't have the next header, so set it to the header length
    streamBuffer_SetWatermark(event, true, false,
                              sizeof(header_control_message), 0);
  } else {
    // set it to the packet length
    streamBuffer_SetWatermark(event, true, false, stream->nextMessageLength, 0);
  }
  parcEventBuffer_Destroy(&input);
}

static void _setConnectionState(_StreamState *stream, bool isUp) {
  parcAssertNotNull(stream, "Parameter stream must be non-null");

  Messenger *messenger = forwarder_GetMessenger(stream->forwarder);

  bool oldStateIsUp = stream->isUp;
  stream->isUp = isUp;

  if (oldStateIsUp && !isUp) {
    // bring connection DOWN
    Missive *missive = missive_Create(MissiveType_ConnectionDown, stream->id);
    messenger_Send(messenger, missive);
    return;
  }

  if (!oldStateIsUp && isUp) {
    // bring connection UP
    Missive *missive = missive_Create(MissiveType_ConnectionUp, stream->id);
    messenger_Send(messenger, missive);
    return;
  }
}

static void _conn_eventcb(PARCEventQueue *event, PARCEventQueueEventType events,
                          void *ioOpsVoid) {
  IoOperations *ops = (IoOperations *)ioOpsVoid;
  _StreamState *stream = (_StreamState *)ioOperations_GetClosure(ops);

  if (events & PARCEventQueueEventType_Connected) {
    if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                          PARCLogLevel_Info)) {
      logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Info, __func__,
                 "Connection %u is connected", stream->id);
    }

    // if the stream was closed, do not transition to an UP state
    if (!stream->isClosed) {
      _setConnectionState(stream, true);
    }
  } else if (events & PARCEventQueueEventType_EOF) {
    if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                          PARCLogLevel_Info)) {
      logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Info, __func__,
                 "connid %u closed.", stream->id);
    }

    parcEventQueue_Disable(stream->bufferEventVector, PARCEventType_Read);

    _setConnectionState(stream, false);

    if (!stream->isClosed) {
      stream->isClosed = true;
      // this will cause the connection manager to destroy the connection later
      messenger_Send(forwarder_GetMessenger(stream->forwarder),
                     missive_Create(MissiveType_ConnectionClosed, stream->id));
    }
  } else if (events & PARCEventQueueEventType_Error) {
    if (logger_IsLoggable(stream->logger, LoggerFacility_IO,
                          PARCLogLevel_Error)) {
      logger_Log(stream->logger, LoggerFacility_IO, PARCLogLevel_Error,
                 __func__, "Got an error on the connection %u: %s", stream->id,
                 strerror(errno));
    }

    parcEventQueue_Disable(stream->bufferEventVector,
                           PARCEventType_Read | PARCEventType_Write);

    _setConnectionState(stream, false);

    if (!stream->isClosed) {
      stream->isClosed = true;
      // this will cause the connection manager to destroy the connection later
      messenger_Send(forwarder_GetMessenger(stream->forwarder),
                     missive_Create(MissiveType_ConnectionClosed, stream->id));
    }
  }
  /* None of the other events can happen here, since we haven't enabled
   * timeouts */
}

static connection_state_t _streamConnection_getState(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->state;
}

static void _streamConnection_setState(IoOperations *ops, connection_state_t state) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  _StreamState *stream =
      (_StreamState *)ioOperations_GetClosure(ops);
  stream->state = state;
}

static connection_state_t _streamConnection_getAdminState(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->admin_state;
}

static void _streamConnection_setAdminState(IoOperations *ops, connection_state_t admin_state) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  _StreamState *stream =
      (_StreamState *)ioOperations_GetClosure(ops);
  stream->admin_state = admin_state;
}

#ifdef WITH_POLICY
static uint32_t _streamConnection_getPriority(const IoOperations *ops) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  const _StreamState *stream =
      (const _StreamState *)ioOperations_GetClosure(ops);
  return stream->priority;
}

static void _streamConnection_setPriority(IoOperations *ops, uint32_t priority) {
  parcAssertNotNull(ops, "Parameter must be non-null");
  _StreamState *stream =
      (_StreamState *)ioOperations_GetClosure(ops);
  stream->priority = priority;
}
#endif /* WITH_POLICY */

static const char * _streamConnection_getInterfaceName(const IoOperations *ops)
{
  parcAssertNotNull(ops, "Parameter must be non-null");
  _StreamState *stream =
      (_StreamState *)ioOperations_GetClosure(ops);
  return stream->interfaceName;
}