net.i2p.client.streaming.impl

## Class MessageInputStream

- java.lang.Object
- - java.io.InputStream
  - - net.i2p.client.streaming.impl.MessageInputStream

- All Implemented Interfaces:
    Closeable, AutoCloseable

      class MessageInputStream
      extends 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 Summary

    Constructors
    Constructor and Description
    `MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize)` Constructs a new MessageInputStream with the specified buffer limits.
    `MessageInputStream(I2PAppContext ctx, int maxMessageSize, int maxWindowSize, int maxBufferSize, int maxPacketCount)` Constructs a new MessageInputStream with the specified buffer limits.

  - ### Method Summary

    All Methods Instance Methods Concrete Methods
    Modifier and Type | Method and 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.
    `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.
    `long[]` | `getNacks()` Gets an array of missing block IDs between the highest ready block and highest received block.
    `int` | `getReadTimeout()` Returns the current read timeout in milliseconds.
    `int` | `getTotalReadySize()` Gets the total number of bytes queued up in ready buffers.
    `boolean` | `isLocallyClosed()` Checks if this stream has been closed locally.
    `boolean` | `messageReceived(long messageId, ByteArray payload)` Accepts a newly received message block.
    `void` | `notifyActivity()` Wakes up any threads waiting for data.
    `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)` Sets the read timeout in milliseconds.
    `(package private) void` | `streamErrorOccurred(IOException ioe)` Records a stream error and marks the stream as locally closed.
    `void` | `updateAcks(PacketLocal packet)` Updates the ACK and NACK fields of a packet based on current stream state.

    - ### Methods inherited from class java.io.InputStream

      `mark, markSupported, reset, skip`

    - ### Methods inherited from class java.lang.Object

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

- - ### Constructor Detail

    - #### MessageInputStream

          public MessageInputStream(I2PAppContext ctx,
                                    int maxMessageSize,
                                    int maxWindowSize,
                                    int maxBufferSize)

      Constructs a new MessageInputStream 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)

      Constructs a new MessageInputStream 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 Detail

    - #### 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

    - #### 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

    - #### 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 `Closeable`
      Specified by:
        `close` in interface `AutoCloseable`
      Overrides:
        `close` in class `InputStream`

    - #### closeReceived

          public void closeReceived()

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

    - #### getHighestBlockId

          public long getHighestBlockId()

      Returns the highest block ID received so far, regardless of order.
      Returns:
        highest block ID received, or -1 if none

    - #### 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

    - #### getNacks

          public long[] getNacks()

      Gets an array of missing block IDs between the highest ready block and highest received block.
      Returns:
        array of missing block IDs, or null if none

    - #### getReadTimeout

          public int getReadTimeout()

      Returns the current read timeout in milliseconds. -1 means block indefinitely, 0 means non-blocking.
      Returns:
        read timeout in ms

    - #### getTotalReadySize

          public int getTotalReadySize()

      Gets the 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

    - #### 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

    - #### notifyActivity

          public void notifyActivity()

      Wakes up any threads waiting for data.

    - #### 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.
      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

    - #### setReadTimeout

          public void setReadTimeout(int timeout)

      Sets the read timeout in milliseconds. -1 means block indefinitely, 0 means non-blocking.
      Parameters:
        `timeout` - new read timeout

    - #### 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

    - #### 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