I/O

Since ProSA is designed to connect to external systems, helpers for I/O operation are essential.

This page covers I/O from server and client perspectives, with a few examples.

Listener

StreamListener type is used to handle server sockets. It can be instantiated from a ListenerSetting using the bind method.

It supports three types of server sockets:

• UNIX
• TCP
• SSL

Once the object is created, you must call the accept method in a loop to accept client connections. Each accepted connection will create a Stream socket, which can be managed just like a client socket.

accept negotiates SSL before it returns, so the loop carries the handshake of every client. A handshake lasts up to the configured ssl_timeout, which is long enough for a slow client to keep the loop from accepting anyone else. An SSL server should instead accept with accept_raw, take an owned snapshot of the SSL parameters with handshaker(), and handshake in a spawned task:

let (stream, addr) = listener.accept_raw().await?;

// The snapshot shares the SSL context, so it costs an atomic increment
let handshaker = listener.handshaker().cloned();
tokio::spawn(async move {
    let stream = match handshaker {
        Some(handshaker) => handshaker.handshake(stream).await?,
        None => stream,
    };

    // ... serve the client
});

The snapshot is taken when the client is accepted, so a configuration reloaded meanwhile applies to the clients accepted afterwards and leaves the handshakes in flight alone. A certificate renewed on disk needs no reload: every snapshot shares the context, which is built again once its files change, as described in the stream settings.

Sharing the listener

set_handshaker takes the listener by value, so a listener that is shared — held in an Arc by the accept loop and by whatever else drives it — cannot be replaced when its configuration changes. Bind it with bind_raw instead: it binds the socket and hands the SSL parameters back beside it rather than inside it, so the handshaker you hold is the only copy and replacing it is an assignment.

let (listener, mut handshaker) = listener_setting.bind_raw().await?;
let listener = Arc::new(listener);

// Accept loop, same shape, from the handshaker held next to the listener
let (stream, addr) = listener.accept_raw().await?;
let client_handshaker = handshaker.clone();
tokio::spawn(async move { /* ... */ });

// Configuration reload or certificate-source change notification
handshaker = listener_setting.build_handshaker().await?;

A listener bound this way carries no SSL parameters of its own: it formats as tcp://, its handshaker() is None, and accept returns a plain TCP stream. That is what keeps a replacement honest — there is no copy left inside the listener for accept, handshake or Debug to serve after you have replaced yours.

Stream

A Stream represents a client socket or a socket created by a StreamListener when a client connects. It can be instantiated from a TargetSetting using the connect method.

It supports several types of client sockets:

• UNIX
• TCP
• SSL
• TCP over HTTP proxy
• SSL over HTTP proxy

You can manually connect using the connect_* methods as appropriate.

After creating the socket, you have several options to further configure it, such as:

• nodelay
• ttl

IO macro

#[io] writes the plumbing every socket object needs. On a struct it injects a stream of a new generic IO type, the remote addr, a read buffer, and a socket_id, and it generates From<IO> and From<(IO, SocketAddr)> so a socket is built straight from a connected Stream. On an impl block it adds the same IO generic, so your methods see the injected fields as if you had declared them. It does not implement the IO trait — parse_frame, read_frame and write_frame describe your protocol and stay yours to write.

#[io]
struct MyIo {
    // Your own fields, which must all implement `Default`
    state: MyState,
}

let (stream, addr) = listener.accept_raw().await?;
let mut socket = MyIo::from((stream, addr));

A socket is built with the file descriptor it happens to hold as its socket_id, which is the queue id it declares itself to the bus with. That value is reused by the system once the socket closes, and is a different one every time the socket reconnects. A processor that keeps a socket across its reconnections should give it an id of its own with set_socket_id, so the bus keeps addressing the same queue:

socket.set_socket_id(slot_id);
proc.add_proc_queue(tx_queue, socket.socket_id()).await?;

Keeping sockets in line with the configuration

A processor that owns sockets has to answer three questions every time its configuration is reloaded: which sockets should still exist, what changed for the ones that stay, and what to do with the requests still addressed to the ones that go.

Deciding which sockets should exist stays with the processor, because only it knows what makes two of its connections the same — a target URL, a pool index, a protocol-level name. The rest is the same work in every protocol, and prosa::io::pool provides it.

A control channel reaches a socket out of band from the requests it serves. A bus queue only reaches a socket that is connected and reading it, so it cannot carry a decision meant for one that is busy serving or waiting to reconnect. Put in it whatever applies to the next attempt or the next message; what defines the connection itself does not belong there, since a socket that has to reach somewhere else is a different socket.

let (control_tx, control_rx) = control_channel(MySocketControl { timeout });

// Configuration reload: the socket picks this up without reconnecting
control_tx.update(|control| control.timeout = new_timeout);

// The slot is gone from the configuration: the socket answers what it has and does not come back
control_tx.stop();

In the socket, stopped() resolves only once the processor retired it, and treats a dropped sender as a retirement — so it is safe as a select! arm, where a plain changed() would spin:

tokio::select! {
    _ = control.stopped() => break,
    Some(msg) = rx_queue.recv() => { /* ... */ }
}

Backoff spaces out reconnection attempts to a peer that is down, doubling from a base delay up to a maximum. A connection that served nothing but lasted at least the base delay counts as a working one and resets the count, so a peer that accepts and closes straight away is not dialled in a loop.

retire_queue drains the queue of a socket that will not come back, answering UnableToReachService. The bus is told to remove the queue, but a processor holding an older service table still sends to it; closing it there would turn a request that should have come back as an error into a send failure for its sender.