connection.rs

Overview

This file implements the Connection abstraction for managing QUIC connections using the msquic library. It provides the ability to create, start, accept, and shutdown connections, as well as open and accept streams, send and receive datagrams, and handle remote certificates. The Connection struct encapsulates an underlying msquic::Connection along with internal state and synchronization primitives to manage asynchronous events and concurrent access.

Key responsibilities include:

• Establishing and starting connections.

• Handling connection lifecycle events and state transitions.

• Managing inbound and outbound streams (bidirectional and unidirectional).

• Sending and receiving datagrams.

• Managing remote peer certificates and ALPN negotiation.

• Providing futures for asynchronous connection operations.

The file handles asynchronous operations through Rust futures and uses wakers to notify tasks of state changes. Internal synchronization is achieved via a Mutex guarding the connection's exclusive state.

Main Structs and Their Responsibilities

Connection

A thread-safe handle representing a QUIC connection. It wraps an Arc to a ConnectionInstance that contains the internal state and the actual msquic::Connection.

Key Methods

• new(registration: &msquic::Registration, credential: NetCredential) -> Result<Self, ConnectionError>
  Creates a new connection instance but does not start it.
  Parameters:

  • registration: Registration handle for msquic.

  • credential: Network credential used for certificate verification.
    Returns:

  • Ok(Connection) on success or Err(ConnectionError) on failure.

• start(configuration: &msquic::Configuration, host: &str, port: u16) -> ConnectionStart
  Starts the connection asynchronously to a specified host and port. Returns a future that completes when the connection is established.
  Usage Example:

  let conn = Connection::new(&registration, credential)?;
  let start_future = conn.start(&configuration, "example.com", 443);
  start_future.await?;
  

• accept() -> ConnectionAccept
  Waits asynchronously for the connection to complete acceptance (e.g., incoming connections). Returns a future that resolves on connection acceptance.

• open_outbound_stream(stream_type: StreamType, fail_on_blocked: bool) -> OpenOutboundStream
  Opens a new outbound stream of the specified type asynchronously. If fail_on_blocked is true, the future will fail if the stream cannot start immediately.

• open_send_stream() -> OpenSendStream
  Opens a new outbound message-based send stream.

• accept_inbound_stream() -> AcceptInboundStream
  Accepts an inbound bidirectional stream asynchronously.

• accept_inbound_uni_stream() -> AcceptInboundUniStream
  Accepts an inbound unidirectional stream asynchronously.

• send_datagram(buf: &Bytes) -> Result<(), DgramSendError>
  Sends a datagram immediately. Returns an error if the connection is not in a proper state.

• get_local_addr() -> Result<SocketAddr, ConnectionError>
  Returns the local socket address for this connection.

• get_remote_addr() -> Result<SocketAddr, ConnectionError>
  Returns the remote socket address for this connection.

• get_remote_certificate() -> Option<CertificateDer<'static>>
  Retrieves the remote certificate if available.

• watch_shutdown() -> WatchShutdown
  Returns a future that completes when the connection shutdown completes.

Other methods provide polling implementations to drive the futures returned by the above asynchronous operations.

Internal Structs

ConnectionInstance

Holds the actual msquic::Connection and an Arc to ConnectionInner. Implements Deref for convenient access to inner data.

ConnectionInner

Holds the shared NetCredential and a Mutex-protected ConnectionInnerExclusive struct that contains the mutable state.

ConnectionInnerExclusive

Manages the mutable connection state, including:

• Current connection state (ConnectionState enum).

• Errors encountered.

• Lists of Wakers for various asynchronous operations.

• Queues of inbound streams.

• Buffers for received datagrams.

• Pool of reusable write buffers.

• Negotiated ALPN bytes.

• Stored remote certificate.

Connection State Machine

The connection state transitions are managed via the ConnectionState enum:

• Open: Newly created but not started.

• Connecting: Connection start initiated, awaiting completion.

• Connected: Connection established.

• Shutdown: Shutdown in progress.

• ShutdownComplete: Shutdown finished.

State transitions are triggered by msquic connection events handled in the callback_handler_impl method of ConnectionInner.

Event Handling and Callbacks

The ConnectionInner::callback_handler_impl method processes various msquic::ConnectionEvents:

• Connected: Marks connection as connected, stores negotiated ALPN, and wakes start waiters.

• ShutdownInitiatedByTransport: Marks shutdown by transport, records error, wakes waiters.

• ShutdownInitiatedByPeer: Marks shutdown by peer, records error, wakes waiters.

• ShutdownComplete: Marks shutdown complete and wakes all waiters.

• PeerStreamStarted: Processes inbound streams (uni- or bidirectional) and notifies waiters.

• StreamsAvailable, DatagramStateChanged, DatagramReceived, and DatagramSendStateChanged: Handle stream availability and datagram state changes.

• PeerCertificateReceived: Validates and stores the remote certificate, wakes certificate waiters.

All event handlers update the internal state and notify waiting futures by waking the appropriate Wakers.

Futures for Asynchronous Operations

Several structs implement Future for asynchronous connection operations:

• ConnectionStart: Future for starting a connection.

• ConnectionAccept: Future for accepting connection establishment.

• RemoteCertificateReceiver: Future for receiving the remote certificate.

• OpenOutboundStream: Future for opening an outbound stream.

• OpenSendStream: Future for opening a message-based send stream.

• AcceptInboundStream: Future for accepting inbound bidirectional streams.

• AcceptInboundUniStream: Future for accepting inbound unidirectional streams.

• WatchShutdown: Future to await connection shutdown.

Each future polls the connection's internal state and either completes immediately or registers its waker to be notified when the awaited event occurs.

Error Types

Several error enums represent different error conditions:

• ConnectionError: Errors related to connection lifecycle and shutdown.

• DgramReceiveError: Errors when receiving datagrams (e.g., connection not started, lost).

• DgramSendError: Errors when sending datagrams (e.g., denied, too big, connection lost).

• StartError: Errors when starting a connection (including certificate validation failures).

• ShutdownError: Errors in connection shutdown process.

Each error type implements detailed error messages using the thiserror crate.

Interaction with Other Components

• Uses msquic for underlying QUIC connection and event handling.

• Works with Stream, ReadStream, and SendStream types from the stream module for stream abstractions.

• Uses NetCredential for certificate verification.

• Employs WriteBuffer for datagram send buffering.

• Uses Bytes crate for datagram data.

• Integrates with rustls_pki_types::CertificateDer for certificate representation.

• Relies on synchronization primitives (Arc, Mutex) and async primitives (Waker, Future) from Rust standard libraries.

This file is central to managing connection-level functionality and serves as a foundation for stream and datagram operations.

Important Implementation Details

• Uses a Mutex-protected exclusive state (ConnectionInnerExclusive) to safely manage mutable state across threads.

• Event callbacks wake all relevant task wakers to ensure futures progress promptly.

• Uses VecDeque queues for inbound streams and datagram buffers to maintain order.

• Implements pooling of WriteBuffer instances to optimize datagram sending.

• Asynchronous operation futures perform polling by inspecting current connection state and registering wakers if the operation cannot proceed immediately.

• Certificate verification is done asynchronously as part of the peer certificate event, with errors stored for later reporting.

• Connection shutdown is carefully handled to notify all waiters and transition states properly.

Visual Diagram

classDiagram
class Connection {
+new()
+start()
+accept()
+open_outbound_stream()
+open_send_stream()
+accept_inbound_stream()
+accept_inbound_uni_stream()
+send_datagram()
+get_local_addr()
+get_remote_addr()
+get_remote_certificate()
+watch_shutdown()
}
class ConnectionInstance {
-inner: Arc<ConnectionInner>
-msquic_conn: msquic::Connection
}
class ConnectionInner {
-credential: NetCredential
-exclusive: Mutex<ConnectionInnerExclusive>
+callback_handler_impl()
}
class ConnectionInnerExclusive {
-state: ConnectionState
-error: Option<ConnectionError>
-start_waiters: Vec<Waker>
-certificate_waiters: Vec<Waker>
-inbound_stream_waiters: Vec<Waker>
-inbound_uni_stream_waiters: Vec<Waker>
-inbound_streams: VecDeque<Stream>
-inbound_uni_streams: VecDeque<ReadStream>
-recv_buffers: VecDeque<Bytes>
-recv_waiters: Vec<Waker>
-write_pool: Vec<WriteBuffer>
-shutdown_waiters: Vec<Waker>
-negotiates_alpn: Vec<u8>
-certificate: Option<Result<CertificateDer, StartError>>
}
Connection "1" o-- "1" ConnectionInstance
ConnectionInstance "1" o-- "1" ConnectionInner
ConnectionInner "1" o-- "1" ConnectionInnerExclusive

Usage Examples

Creating and Starting a Connection

let registration = msquic::Registration::new("myapp")?;
let credential = NetCredential::new();
let conn = Connection::new(&registration, credential)?;
let configuration = msquic::Configuration::load_default()?;

conn.start(&configuration, "example.com", 443).await?;

Accepting an Inbound Bidirectional Stream

let stream = conn.accept_inbound_stream().await?;

Sending a Datagram

let data = Bytes::from("hello datagram");
conn.send_datagram(&data)?;

Receiving a Remote Certificate

let cert = conn.receive_remote_certificate().await?;

References

• See Stream for stream-related details.

• See NetCredential for credential and certificate verification logic.

• See msquic for underlying QUIC connection and event definitions.

• See CertificateDer for certificate representation.

• See WriteBuffer for datagram buffer management.

The asynchronous patterns and futures conform to Rust's async runtime model, detailed in Asynchronous Programming.