java.lang.Object
java.io.InputStream
net.i2p.client.streaming.impl.MessageInputStream
All Implemented Interfaces:
Closeable, AutoCloseable
class MessageInputStreamextends InputStream
InputStream implementation that accepts messages arriving out of order and presents the data in proper sequence for reading.

This class buffers incoming data blocks and reorders them transparently, ensuring a continuous byte stream to the caller. It supports:

  • Out-of-order message handling
  • Flow control via blocking reads
  • Optional read timeouts
  • EOF signaling via closeReceived()
  • Error propagation via streamErrorOccurred()

Thread Safety: This class is synchronized internally and safe for use by one reader thread and potentially multiple message-receiving threads.

Performance: Designed for high-throughput scenarios such as large file downloads. Avoids unnecessary copying by using ByteArray references.

Lifecycle: The stream can be closed locally or receive an EOF signal remotely. Once closed, no further reads or message receptions are allowed.

Usage:

I2PSession -> MessageHandler -> PacketHandler -> ConnectionPacketHandler -> MessageInputStream

  • Constructor Details

    • MessageInputStream

      public MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize)
      New stream with the specified buffer limits.
      Parameters:
      ctx - Application context for logging and utilities
      maxMessageSize - Max size of a single message block
      maxWindowSize - Max number of messages in the window
      maxBufferSize - Max total size of all buffered data

    • MessageInputStream

      public MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize, int maxPacketCount)
      New stream with the specified buffer limits.
      Parameters:
      ctx - Application context for logging and utilities
      maxMessageSize - Max size of a single message block
      maxWindowSize - Max number of messages in the window
      maxBufferSize - Max total size of all buffered data
      maxPacketCount - Max number of packets regardless of byte size

  • Method Details

    • getHighestReadyBlockId

      public long getHighestReadyBlockId()
      Returns the highest consecutive block ID that has been fully received and marked ready.
      Returns:
      highest ready block ID, or -1 if none

    • getHighestBlockId

      public long getHighestBlockId()
      Returns the highest block ID received so far, regardless of order.
      Returns:
      highest block ID received, or -1 if none

    • getTotalReadySize

      public int getTotalReadySize()
      Total number of bytes queued up in ready buffers. Does not throw IOException on closed stream.
      Returns:
      number of bytes waiting to be read

    • isLocallyClosed

      public boolean isLocallyClosed()
      Checks if this stream has been closed locally.
      Returns:
      true if closed, false otherwise

    • canAccept

      public boolean canAccept(long messageId, int payloadSize)
      Determines whether the stream can accept a new message.

      Accepts zero-length or duplicate packets always.

      Parameters:
      messageId - The sequence ID of the message
      payloadSize - Size of the message payload
      Returns:
      true if the message can be accepted, false otherwise

    • growBuffer

      public boolean growBuffer()
      Try to grow the receive buffer when canAccept() returns false. Doubles the buffer size up to a reasonable cap (8MB) to reduce choke frequency for high-throughput connections.
      Returns:
      true if the buffer was grown, false if already at cap

    • ensureCanAccept

      public boolean ensureCanAccept(long messageId, int payloadSize)
      Combined canAccept with automatic buffer growth. If the buffer is full, tries to grow it before rejecting.
      Parameters:
      messageId - the inbound message being offered
      payloadSize - the size in bytes the message needs
      Returns:
      true if the message can be accepted (possibly after buffer growth)

    • getNacks

      public long[] getNacks()
      Missing block IDs between the highest ready block and highest received block. Two-pass to avoid ArrayList allocation inside the lock.
      Returns:
      array of missing block IDs, empty if none

    • getNacksLocked

      long[] getNacksLocked()
      Returns missing block IDs. Must be called while holding _dataLock.
      Returns:
      array of missing block IDs, empty if none

    • updateAcks

      public void updateAcks(PacketLocal packet)
      Updates the ACK and NACK fields of a packet based on current stream state.
      Parameters:
      packet - Packet to update

    • closeReceived

      public void closeReceived()
      Notifies the stream that no more messages will be received after the current highest. Buffered data remains accessible until read.

    • messageReceived

      public boolean messageReceived(long messageId, ByteArray payload)
      Accepts a newly received message block.

      If the message is next in sequence, it's added to the ready list and contiguous out-of-order messages are promoted.

      Parameters:
      messageId - The message sequence ID
      payload - The message payload
      Returns:
      true if the message was new, false if a duplicate

    • read

      public int read() throws IOException
      Reads a single byte from the stream.
      Specified by:
      read in class InputStream
      Returns:
      the next byte (0-255), or -1 on EOF
      Throws:
      IOException - if stream is closed or read fails

    • read

      public int read(byte[] target) throws IOException
      Reads bytes into the given array. Throws SocketTimeoutException on timeout.
      Overrides:
      read in class InputStream
      Parameters:
      target - byte array to read into
      Returns:
      number of bytes read, or -1 on EOF
      Throws:
      IOException - if stream is closed or read fails

    • read

      public int read(byte[] target, int offset, int length) throws IOException
      Reads up to length bytes into the target array at offset. Blocks until data is available or timeout occurs.

      The lock is held only during state checks, waits, and index updates. The actual System.arraycopy(Object, int, Object, int, int) is performed outside the synchronized block to reduce contention with the receive thread (messageReceived(long, ByteArray)). This is safe because ByteArray payloads are immutable once stored.

      Overrides:
      read in class InputStream
      Parameters:
      target - byte array to read into
      offset - offset in array to start writing
      length - number of bytes to read
      Returns:
      number of bytes read, or -1 on EOF
      Throws:
      IOException - if stream is closed or read fails

    • available

      public int available() throws IOException
      Returns an estimate of bytes readily available to read without blocking.
      Overrides:
      available in class InputStream
      Returns:
      number of bytes available for reading
      Throws:
      IOException - if stream is closed or error occurred

    • getReadTimeout

      public int getReadTimeout()
      Returns the current read timeout in milliseconds. -1 means block indefinitely, 0 means non-blocking.
      Returns:
      read timeout in ms

    • setReadTimeout

      public void setReadTimeout(int timeout)
      Read timeout in milliseconds. -1 means block indefinitely, 0 means non-blocking.
      Parameters:
      timeout - new read timeout

    • close

      public void close()
      Closes the stream, clears all buffered data, and marks the stream as locally closed. Subsequent reads will throw IOException.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Overrides:
      close in class InputStream

    • streamErrorOccurred

      void streamErrorOccurred(IOException ioe)
      Records a stream error and marks the stream as locally closed. Wakes up any waiting threads.
      Parameters:
      ioe - The IOException that occurred