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));
    }
