rust-yamux

Yamux multiplexes independent I/O streams over a single reliable, ordered connection (like TCP/IP). The library provides two primary abstractions:

1. Connection: Wraps the underlying I/O resource (e.g., a socket). It is used to manage the connection lifecycle and provides methods to either open new outbound streams or accept incoming inbound streams.
2. Stream: Represents an individual multiplexed stream. It implements futures::io::AsyncRead and futures::io::AsyncWrite, allowing you to treat a multiplexed stream like any other asynchronous I/O object.

A Connection moves through several internal states that affect how it responds to polling:

• Active: The connection is healthy and actively multiplexing streams.
• Closing: A graceful shutdown has been requested; the connection is working on finishing pending tasks.
• Cleanup: An unrecoverable error occurred, and the connection is cleaning up resources.
• Closed: The connection is no longer active and cannot be reused.
• Poisoned: An internal error occurred during state transitions (should not be reachable by users).

A Yamux Stream moves through several states that determine its capabilities. You can check the current state using is_write_closed(), is_closed(), and is_pending_ack().

Stream States:

• Open { acknowledged: bool }: The stream is active. For outbound streams, acknowledged tracks if the remote has acknowledged our stream. For inbound streams, it tracks if we have acknowledged the stream to the remote.
• SendClosed: The stream is closed for outgoing messages but remains open for incoming messages.
• RecvClosed: The stream is closed for incoming messages but remains open for outgoing messages.
• Closed: The terminal state; no further reading or writing is possible.

Use can_read() and can_write() to check if operations are permitted in the current state.

The send_on_single_stream function takes a single yamux::Stream and an iterator of Msg objects, sending each message sequentially on that specific stream before closing it.

send_on_single_stream(stream, messages).await?;

The Stream struct provides standard asynchronous I/O interfaces for multiplexed data transfer. It can be used in two ways:

1. As AsyncRead and AsyncWrite: Use poll_read and poll_write to interact with the stream using a byte buffer. This is the standard way to handle raw byte streams.
2. As futures::stream::Stream: Use poll_next to receive data as Packet objects. Each Packet contains a Vec<u8> representing a chunk of data received from the remote.

Note that Stream handles flow control (window updates) automatically during these operations.

The set_max_connection_receive_window method sets the upper limit for the total receive window size across all streams in a connection.

Constraints:

• The value must be >= 256 KiB * max_num_streams to ensure every stream can at least accommodate the Yamux default window size.
• Setting it to None disables the limit, allowing streams to grow their receive windows based on bandwidth-delay-product without restriction.

Security Note: Setting a reasonable limit helps mitigate DOS attacks where an attacker attempts to exhaust memory by forcing the allocation of large stream receive windows.

You can customize the behavior of a Yamux connection using the Config struct. The default configuration is:

• max_connection_receive_window: Some(1 GiB)
• max_num_streams: 512
• read_after_close: true
• split_send_size: 16 KiB

Use the builder-style methods on Config to adjust these values before initializing your connection.

let mut config = Config::default();
config
    .set_max_num_streams(100)
    .set_split_send_size(32 * 1024)
    .set_read_after_close(false);

The echo_server function takes a yamux::Connection and, for every incoming stream, echoes the received data back to the sender. It handles concurrent streams automatically.

echo_server(connection).await?;