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

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    private boolean
    _closeReceived
    Whether an EOF signal has been received from the remote peer.
    private final int
    _configuredBufferSize
    The size this stream was configured with.
    private final Object
    _dataLock
    Lock for the synchronized buffer access.
    private long
    _highestBlockId
    ID of the highest block received (including out-of-order).
    private long
    _highestReadyBlockId
    ID of the highest block fully received in order.
    private final AtomicBoolean
    _locallyClosed
    Whether this stream has been closed locally for reading.
    private final Log
    _log
    Logger for the message input stream.
    private int
    _maxBufferSize
    Largest total buffer size in bytes.
    private final int
    _maxMessageSize
    Largest size of one message block.
    private final int
    _maxPacketCount
    Most packets held regardless of size.
    private final int
    _maxWindowSize
    Most messages in the receive window.
    private final Map<Long, ByteArray>
    _notYetReadyBlocks
    Blocks received out of order, keyed by block ID.
    private final byte[]
    _oneByte
    Buffer for single byte reads.
    private int
    _peakBuffered
    Most bytes buffered since the last decay check.
    private int
    _readTimeout
    Read timeout in milliseconds before returning -1.
    private long
    _readTotal
    Total number of bytes read from this stream.
    private int
    _readyDataBlockIndex
    Current read position within the active ready block.
    private final List<ByteArray>
    _readyDataBlocks
    Ordered list of fully received data blocks ready for reading.
    private int
    _readyDataSize
    Total bytes of the ready data blocks, for a fast available and canAccept.
    private IOException
    _streamError
    Error that caused the stream to fail, or null if healthy.
    private static final ByteArray
    DUMMY_BA
    Sentinel used for out-of-order duplicates when stream is closed.
    private static final int
    INITIAL_CAPACITY
    Initial capacity for internal lists and maps.
    private static final int
    MAX_GROW_BUFFER
    Maximum size the receive buffer can grow to via dynamic growth.
    private static final int
    MAX_PACKET_COUNT
    Maximum number of packets to buffer regardless of byte size.
    private static final int
    MIN_READY_BUFFERS
    Minimum number of ready buffers before limiting applies.

  • Constructor Summary

    Constructors
    Constructor
    Description
    MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize)
    New stream with the specified buffer limits.
    MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize, int maxPacketCount)
    New stream with the specified buffer limits.

  • Method Summary

    Modifier and Type
    Method
    Description
    int
    available()
    Returns an estimate of bytes readily available to read without blocking.
    boolean
    canAccept(long messageId, int payloadSize)
    Determines whether the stream can accept a new message.
    void
    close()
    Closes the stream, clears all buffered data, and marks the stream as locally closed.
    void
    closeReceived()
    Notifies the stream that no more messages will be received after the current highest.
    private void
    decayBuffer()
    Decay the receive-buffer cap toward the configured size after the reader empties the buffer.
    (package private) static int
    decayBufferSize(int current, int floor, int peak)
    Decide the receive-buffer cap once the reader has emptied the buffer.
    boolean
    ensureCanAccept(long messageId, int payloadSize)
    Combined canAccept with automatic buffer growth.
    long
    getHighestBlockId()
    Returns the highest block ID received so far, regardless of order.
    long
    getHighestReadyBlockId()
    Returns the highest consecutive block ID that has been fully received and marked ready.
    (package private) int
    getMaxBufferSize()
    Current receive-buffer cap, including any growth done by growBuffer().
    long[]
    getNacks()
    Missing block IDs between the highest ready block and highest received block.
    (package private) long[]
    getNacksLocked()
    Returns missing block IDs.
    int
    getReadTimeout()
    Returns the current read timeout in milliseconds.
    int
    getTotalReadySize()
    Total number of bytes queued up in ready buffers.
    boolean
    growBuffer()
    Try to grow the receive buffer when canAccept() returns false.
    boolean
    isLocallyClosed()
    Checks if this stream has been closed locally.
    private void
    logBufferBlockWindowExceeded(long messageId)
    Logs a block beyond the receive window.
    private void
    logBufferFull(long messageId, int available)
    Logs a message that did not fit the buffer.
    private void
    logTooManyReadyBlocks(long messageId)
    Logs a block held back by too many ready blocks.
    boolean
    messageReceived(long messageId, ByteArray payload)
    Accepts a newly received message block.
    private void
    noteBuffered()
    Record the buffered level for the decay check.
    int
    read()
    Reads a single byte from the stream.
    int
    read(byte[] target)
    Reads bytes into the given array.
    int
    read(byte[] target, int offset, int length)
    Reads up to length bytes into the target array at offset.
    void
    setReadTimeout(int timeout)
    Read timeout in milliseconds.
    (package private) void
    streamErrorOccurred(IOException ioe)
    Records a stream error and marks the stream as locally closed.
    private void
    throwAnyError()
    Throws any stored stream error and clears it.
    void
    updateAcks(PacketLocal packet)
    Updates the ACK and NACK fields of a packet based on current stream state.

    Methods inherited from class InputStream

    mark, markSupported, nullInputStream, readAllBytes, readNBytes, readNBytes, reset, skip, skipNBytes, transferTo

    Methods inherited from class Object

    clone, equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • INITIAL_CAPACITY

      private static final int INITIAL_CAPACITY
      Initial capacity for internal lists and maps. Helps avoid frequent reallocations for common use cases.
      See Also:

    • MIN_READY_BUFFERS

      private static final int MIN_READY_BUFFERS
      Minimum number of ready buffers before limiting applies. Helps prevent unnecessary buffering limits on slow systems.
      See Also:

    • MAX_PACKET_COUNT

      private static final int MAX_PACKET_COUNT
      Maximum number of packets to buffer regardless of byte size. Prevents excessive memory usage with small packets while allowing higher packet counts for better throughput.
      See Also:

    • _log

      private final Log _log
      Logger for the message input stream.

    • _readyDataBlocks

      private final List<ByteArray> _readyDataBlocks
      Ordered list of fully received data blocks ready for reading.

    • _readyDataBlockIndex

      private int _readyDataBlockIndex
      Current read position within the active ready block.

    • _highestReadyBlockId

      private long _highestReadyBlockId
      ID of the highest block fully received in order.

    • _highestBlockId

      private long _highestBlockId
      ID of the highest block received (including out-of-order).

    • _notYetReadyBlocks

      private final Map<Long, ByteArray> _notYetReadyBlocks
      Blocks received out of order, keyed by block ID.

    • _closeReceived

      private boolean _closeReceived
      Whether an EOF signal has been received from the remote peer.

    • _locallyClosed

      private final AtomicBoolean _locallyClosed
      Whether this stream has been closed locally for reading.

    • _readTimeout

      private volatile int _readTimeout
      Read timeout in milliseconds before returning -1.

    • _streamError

      private IOException _streamError
      Error that caused the stream to fail, or null if healthy.

    • _readTotal

      private long _readTotal
      Total number of bytes read from this stream.

    • _maxMessageSize

      private final int _maxMessageSize
      Largest size of one message block.

    • _maxWindowSize

      private final int _maxWindowSize
      Most messages in the receive window.

    • MAX_GROW_BUFFER

      private static final int MAX_GROW_BUFFER
      Maximum size the receive buffer can grow to via dynamic growth. Matches ConnectionOptions default cap.
      See Also:

    • _maxBufferSize

      private int _maxBufferSize
      Largest total buffer size in bytes.

    • _configuredBufferSize

      private final int _configuredBufferSize
      The size this stream was configured with. Dynamic growth goes above it, the decay never goes below it - the configured size is derived from the window and the message size, so a smaller buffer would drop data the window is able to deliver.

    • _peakBuffered

      private int _peakBuffered
      Most bytes buffered since the last decay check. The peak, not the current level, is what shows whether a grown buffer is actually in use.

    • _maxPacketCount

      private final int _maxPacketCount
      Most packets held regardless of size.

    • _oneByte

      private final byte[] _oneByte
      Buffer for single byte reads.

    • _dataLock

      private final Object _dataLock
      Lock for the synchronized buffer access.

    • DUMMY_BA

      private static final ByteArray DUMMY_BA
      Sentinel used for out-of-order duplicates when stream is closed. When locallyClosed is true and an out-of-order message arrives, we store this sentinel instead of the actual payload to prevent memory accumulation while still allowing canAccept(long, int) to return true for retransmits of already-processed messages. This enables proper handling of late-arriving duplicate packets after stream closure.

      Note: the payload size of this sentinel is 0, so it does not count toward _readyDataSize or buffer limits. See canAccept(long, int).

    • _readyDataSize

      private int _readyDataSize
      Total bytes of the ready data blocks, for a fast available and canAccept.

  • 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

    • decayBufferSize

      static int decayBufferSize(int current, int floor, int peak)
      Decide the receive-buffer cap once the reader has emptied the buffer.

      The peak, not the current level, is the signal: it says how much buffer the drain just finished actually needed. Keeping room for twice the peak means a connection that is genuinely using the larger buffer keeps it, and the cap converges on the working set instead of alternating a grow with a shrink on every burst.

      Parameters:
      current - the cap in effect, in bytes
      floor - the configured size, which the cap never falls below
      peak - most bytes buffered since the previous check, in bytes
      Returns:
      the new cap: half of current clamped to floor, or current if the peak shows the larger buffer is in use

    • decayBuffer

      private void decayBuffer()
      Decay the receive-buffer cap toward the configured size after the reader empties the buffer.

      Without this the cap only ratchets up. One transient burst that trips canAccept() doubles it, and every later packet on that connection is then judged against the larger window for the rest of its life, so the memory stays committed whether the traffic needs it or not.

      Caller must hold _dataLock.

    • noteBuffered

      private void noteBuffered()
      Record the buffered level for the decay check.

      Caller must hold _dataLock.

    • getMaxBufferSize

      int getMaxBufferSize()
      Current receive-buffer cap, including any growth done by growBuffer().
      Returns:
      the cap in bytes

    • 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)

    • logBufferFull

      private void logBufferFull(long messageId, int available)
      Logs a message that did not fit the buffer.
      Parameters:
      messageId - the message that did not fit
      available - bytes still available in the buffer

    • logBufferBlockWindowExceeded

      private void logBufferBlockWindowExceeded(long messageId)
      Logs a block beyond the receive window.
      Parameters:
      messageId - the message that was too far ahead

    • logTooManyReadyBlocks

      private void logTooManyReadyBlocks(long messageId)
      Logs a block held back by too many ready blocks.
      Parameters:
      messageId - the message that was held back

    • 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

    • throwAnyError

      private void throwAnyError() throws IOException
      Throws any stored stream error and clears it.
      Throws:
      IOException - if a stream error has occurred