java.lang.Object
java.io.OutputStream
net.i2p.client.streaming.impl.MessageOutputStream
All Implemented Interfaces:
Closeable, Flushable, AutoCloseable
class MessageOutputStreamextends OutputStream
An OutputStream implementation that buffers data and forwards it to a MessageOutputStream.DataReceiver in chunks upon buffer full or flush. Provides throttling and blocking behavior according to the data receiver's needs.

This class maintains an internal buffer with configurable size limits, supports resizing the buffer size during operation (only effective when buffer is empty), and allows passive flushes after inactivity delays to avoid excessive buffering.

MessageOutputStream -> ConnectionDataReceiver -> Connection -> PacketQueue -> I2PSession

Thread safety: - Write operations synchronize on an internal data lock when manipulating the buffer. - Waiting for buffer acceptance and completion is done outside synchronization to avoid blocking other threads.

Performance: - Uses a shared ByteCache for buffer byte arrays to reduce allocations. - Supports a passive flush mechanism to flush buffered data after inactivity. - Minimizes locking time by doing blocking waits outside of synchronized blocks.

Note: - Buffer resizing only affects the max size used to determine flush boundaries, buffer reallocation does not occur dynamically. - Stream errors are captured and thrown upon subsequent I/O calls. - Interruptions during wait throw InterruptedIOException accordingly. - Close() flushes any remaining data and releases internal buffers.

  • Field Details

    • _context

      private final I2PAppContext _context
      The app context.

    • _log

      private final Log _log
      Logger for the output stream.

    • _buf

      private byte[] _buf
      The buffer accumulating the written data.

    • _valid

      private int _valid
      Valid bytes held in the buffer.

    • _dataLock

      private final Object _dataLock
      Lock for the write buffer.

    • _oneByte

      private final byte[] _oneByte
      Reusable buffer for single byte writes, guarded by _oneByteLock, not _dataLock.

    • _oneByteLock

      private final Object _oneByteLock
      Lock for the single byte writes.

    • _dataReceiver

      private final MessageOutputStream.DataReceiver _dataReceiver
      The connection receiving the data.

    • _streamError

      private final AtomicReference<IOException> _streamError
      The stream error, kept to rethrow on the next call.

    • _closed

      private final AtomicBoolean _closed
      True once the stream is closed.

    • _written

      private long _written
      Total bytes written to the connection since creation.

    • _writeTimeout

      private volatile int _writeTimeout
      Blocking write timeout, in ms, -1 for no timeout.

    • _dataCache

      private final ByteCache _dataCache
      Cache supplying the write buffer.

    • _originalBufferSize

      private final int _originalBufferSize
      Buffer size requested at construction, the two sizes control when the buffer is flushed.

    • _currentBufferSize

      private int _currentBufferSize
      Buffer size in use now.

    • _nextBufferSize

      private volatile int _nextBufferSize
      Buffer size requested by the next flush, updated when buffer drains.

    • _flusher

      private final MessageOutputStream.Flusher _flusher
      Flusher that sends buffered data after a period of inactivity.

    • _lastBuffered

      private volatile long _lastBuffered
      Timestamp of the most recently buffered data, in context clock ms.

    • _passiveFlushDelay

      private volatile int _passiveFlushDelay
      Delay in milliseconds before a passive flush is triggered.

    • defaultPassiveFlushDelay

      private static volatile int defaultPassiveFlushDelay
      Default passive flush delay optimized for lower latency while maintaining stability: 200 ms on a slow system, 100 ms otherwise.

  • Constructor Details

    • MessageOutputStream

      public MessageOutputStream(I2PAppContext ctx, SimpleTimer2 timer, MessageOutputStream.DataReceiver receiver, int bufSize, int initBufSize)
      Stream with default passive flush delay.
      Parameters:
      ctx - Application context for logging and utilities
      timer - Timer used for scheduling passive flush events
      receiver - DataReceiver that handles actual data dispatch
      bufSize - Maximum buffer size allowed (will be clamped to preset max values)
      initBufSize - Initial buffer size to use for buffering (<= bufSize)

    • MessageOutputStream

      public MessageOutputStream(I2PAppContext ctx, SimpleTimer2 timer, MessageOutputStream.DataReceiver receiver, int bufSize, int initBufSize, int passiveFlushDelay)
      Stream with specified passive flush delay.
      Parameters:
      ctx - Application context for logging and utilities
      timer - Timer used for scheduling passive flush events
      receiver - DataReceiver that handles actual data dispatch
      bufSize - Maximum buffer size allowed (will be clamped for safety)
      initBufSize - Initial buffer size to use for buffering (<= bufSize)
      passiveFlushDelay - Delay in ms before passive flush if data is buffered but not flushed

  • Method Details

    • getDefaultPassiveFlushDelay

      public static int getDefaultPassiveFlushDelay()
      Read the passive flush delay applied when a caller does not name one.
      Returns:
      the flush delay in milliseconds used when none is specified

    • setDefaultPassiveFlushDelay

      public static void setDefaultPassiveFlushDelay(int val)
      Set the passive flush delay used when a caller does not name one.
      Parameters:
      val - requested flush delay in milliseconds, clamped to the range 10 - 500

    • setWriteTimeout

      public void setWriteTimeout(int ms)
      Write timeout in milliseconds for blocking write operations.
      Parameters:
      ms - Timeout in milliseconds, -1 for infinite wait.

    • getWriteTimeout

      public int getWriteTimeout()
      Current write timeout.
      Returns:
      write timeout in milliseconds

    • getPassiveFlushDelay

      public int getPassiveFlushDelay()
      Current passive flush delay.
      Returns:
      current passive flush delay in ms

    • setPassiveFlushDelay

      public void setPassiveFlushDelay(int delayMs)
      Dynamically adjust the passive flush delay. Clamped to [10, 500]. Used by the interactive profile to reduce flush latency mid-connection.
      Parameters:
      delayMs - new delay, will be clamped to valid range

    • setBufferSize

      public void setBufferSize(int size)
      Requests resizing the buffer to the specified size. Effective only when the internal buffer is empty, and must be > 0 and <= the original max buffer size. Does not immediately resize the underlying byte array.
      Parameters:
      size - New desired buffer size, must be positive and no greater than original buffer size.

    • write

      public void write(byte[] b) throws IOException
      Writes the specified byte array to the stream, may block waiting for acceptance.
      Overrides:
      write in class OutputStream
      Parameters:
      b - byte array to write
      Throws:
      IOException - if the stream is closed or write fails

    • write

      public void write(byte[] b, int off, int len) throws IOException
      Writes len bytes from the specified byte array starting at offset off. Blocks if buffer becomes full waiting for acceptance.
      Overrides:
      write in class OutputStream
      Parameters:
      b - byte array source
      off - start offset in array
      len - number of bytes to write
      Throws:
      IOException - if the stream is closed or write fails

    • write

      public void write(int b) throws IOException
      Writes a single byte to the stream.
      Specified by:
      write in class OutputStream
      Parameters:
      b - byte to write (only least significant 8 bits used)
      Throws:
      IOException - if the stream is closed or write fails

    • locked_updateBufferSize

      private final int locked_updateBufferSize()
      Updates the current buffer size if a new size has been requested. This occurs only when the buffer is empty to avoid data loss.
      Returns:
      the effective current buffer size

    • flush

      public void flush() throws IOException
      Flushes buffered data, blocking until accepted into the send queue. Does not wait for ACK from far end, to avoid performance bottlenecks.
      Specified by:
      flush in interface Flushable
      Overrides:
      flush in class OutputStream
      Throws:
      IOException - if flushing encounters an error

    • flush

      private void flush(boolean waitForAcceptOnly) throws IOException
      Flush implementation allowing control over wait behavior.
      Parameters:
      waitForAcceptOnly - true to wait only for buffer acceptance, false to wait for full completion (ACK)
      Throws:
      IOException - if flushing fails or is interrupted

    • close

      public void close() throws IOException
      Closes the stream, flushing all buffered data and waiting for any CLOSE packet acknowledgments.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Overrides:
      close in class OutputStream
      Throws:
      IOException - if closing encounters an error

    • closeInternal

      void closeInternal()
      Non-blocking internal close used within the package. Does not wait for flush or acknowledgment. Marks the stream as closed and pushes any buffered data out fire-and-forget, so a connection closed right after a write does not lose the application's last bytes.

    • clearData

      private void clearData(boolean shouldFlush)
      Clears any buffered data, optionally flushing it out first.
      Parameters:
      shouldFlush - if true, send the buffered data (but do not wait for acknowledgment); if false, discard it. The error path discards, the close path flushes.

    • getClosed

      public boolean getClosed()
      Returns true if the stream is closed.
      Returns:
      true if closed

    • throwAnyError

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

    • streamErrorOccurred

      void streamErrorOccurred(IOException ioe)
      Records a stream error and clears buffered data.
      Parameters:
      ioe - The IOException to record

    • flushAvailable

      void flushAvailable(MessageOutputStream.DataReceiver target, boolean blocking) throws IOException
      Attempts to flush buffered data to the target DataReceiver, optionally blocking until accepted.
      Parameters:
      target - the destination DataReceiver
      blocking - whether the flush should block until accepted
      Throws:
      IOException - if the flush fails or is interrupted

    • flushAvailable

      void flushAvailable(MessageOutputStream.DataReceiver target) throws IOException
      Convenience overload variant with blocking flush.
      Parameters:
      target - DataReceiver to flush to
      Throws:
      IOException - if flush fails or interrupted

    • destroy

      void destroy()
      Releases resources without flushing.