Example: counting over channels

The client asks the server to count, includes a channel sender in the request and reads the resulting sequence from the receiver.

The client creates the channel after establishing the TCP connection. Its sender is transferred to the server as part of the request, while the client retains the receiver. Remoc carries the channel over the existing connection.

The example uses three crates. counting defines the request type, and the client and server binaries both depend on it.

Running it

Clone the repository: 

git clone https://github.com/remoc-rs/remoc
cd remoc

Start the server: 

cargo run --manifest-path examples/channels/Cargo.toml -p counting-server

Then, in another terminal, start the client: 

cargo run --manifest-path examples/channels/Cargo.toml -p counting-client

The program

The shared crate defines the request, and the other two crates provide the server and client binaries.

counting

Defines the request type shared by the client and server.

counting/Cargo.toml

Its dependencies.

[package]
name = "counting"
version = "0.1.0"
edition = "2024"

[dependencies]
remoc = { version = "0.20" }
serde = { version = "1.0", features = ["derive"] }

counting/src/lib.rs

Defines a request containing a channel sender.

//! This library crate defines the data exchanged by the counting example.
//!
//! The client and server depend on it. It is the whole contract between
//! them: there is no schema file and nothing to generate.
#![warn(missing_docs)]

use remoc::prelude::*;

/// TCP port the server is listening on.
pub const TCP_PORT: u16 = 9870;

/// A request to count up to a number.
///
/// Remoc types such as channel senders and receivers are serializable, so they
/// can be placed in a struct like any other field. Sending this request creates
/// the channel `seq_tx` belongs to inside the connection that carries the
/// request, which is what saves the client from opening a second connection or
/// agreeing on an identifier with the server.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct CountReq {
    /// Count up to, but excluding, this number.
    pub up_to: u32,
    /// The sender the server should count into.
    ///
    /// The client keeps the matching receiver.
    pub seq_tx: rch::mpsc::Sender<u32>,
}

counting-server

Accepts TCP connections and counts into whatever channel arrives in the request.

counting-server/Cargo.toml

Its dependencies.

[package]
name = "counting-server"
version = "0.1.0"
edition = "2024"

[dependencies]
counting = { path = "../counting" }
remoc = { version = "0.20" }
tokio = { version = "1", features = ["rt-multi-thread", "net", "time"] }

counting-server/src/main.rs

Receives requests and sends each sequence over the channel included in the request.

//! This crate implements the server of the counting example.
//!
//! It accepts TCP connections, establishes a Remoc connection over each one and
//! then counts into whatever channel the client sends it.
#![warn(missing_docs)]

use remoc::prelude::*;
use std::{net::Ipv4Addr, time::Duration};
use tokio::{net::TcpListener, time::sleep};

use counting::{CountReq, TCP_PORT};

#[tokio::main]
async fn main() {
    // Listen to TCP connections using Tokio.
    // In reality you would probably use TLS or WebSockets over HTTPS.
    println!("Listening on port {TCP_PORT}. Press Ctrl+C to exit.");
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, TCP_PORT)).await.unwrap();

    loop {
        // Accept an incoming TCP connection.
        let (socket, addr) = listener.accept().await.unwrap();
        let (socket_rx, socket_tx) = socket.into_split();
        println!("Accepted connection from {addr}");

        // Spawn a task for each incoming connection.
        tokio::spawn(async move {
            // Establish a Remoc connection with default configuration over the TCP
            // connection and obtain the receiving half of the base channel.
            //
            // The connection is always bidirectional, but this end only receives,
            // so the unneeded sender is dropped.
            let (conn, _tx, rx): (_, rch::base::Sender<()>, _) =
                remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).await.unwrap();

            // The connection dispatcher must be spawned; it drives everything that
            // travels over this TCP connection.
            tokio::spawn(conn);

            // Serve requests until the client disconnects. A client going away is
            // an ordinary end to the conversation, not a failure, so it is told
            // apart from the errors that are.
            match serve(rx).await {
                Ok(()) => println!("Client {addr} closed the connection"),
                Err(err) if err.is_disconnected() => println!("Client {addr} disconnected"),
                Err(err) => println!("Connection from {addr} failed: {err}"),
            }
        });
    }
}

/// Counts for the client, once per request.
async fn serve(mut rx: rch::base::Receiver<CountReq>) -> Result<(), rch::base::RecvError> {
    // Receive count requests over the base channel.
    while let Some(CountReq { up_to, seq_tx }) = rx.recv().await? {
        println!("Counting up to {up_to}");

        for i in 0..up_to {
            // Send each counted number over the channel the client provided.
            //
            // This is an ordinary channel send; that the receiver is on another
            // machine makes no difference to the code, only to the error it can
            // return.
            if seq_tx.send(i).await.into_disconnected().unwrap() {
                // The client dropped its receiver or went away, so there is
                // nobody left to count for.
                println!("Client stopped listening");
                break;
            }

            sleep(Duration::from_millis(300)).await;
        }

        // Dropping seq_tx closes that channel, which is how the client's
        // recv() learns that the sequence has ended.
    }

    Ok(())
}

counting-client

Asks the server to count and prints the sequence as it arrives.

counting-client/Cargo.toml

Its dependencies.

[package]
name = "counting-client"
version = "0.1.0"
edition = "2024"

[dependencies]
counting = { path = "../counting" }
remoc = { version = "0.20" }
tokio = { version = "1", features = ["rt-multi-thread", "net", "time"] }

counting-client/src/main.rs

Creates a channel, sends its sender in the request and reads from its receiver.

//! This crate implements the client of the counting example.
//!
//! It asks the server to count, sending the channel to count into along with
//! the request.
#![warn(missing_docs)]

use remoc::prelude::*;
use std::net::Ipv4Addr;
use tokio::net::TcpStream;

use counting::{CountReq, TCP_PORT};

#[tokio::main]
async fn main() {
    // Establish TCP connection to server.
    let socket = TcpStream::connect((Ipv4Addr::LOCALHOST, TCP_PORT)).await.unwrap();
    let (socket_rx, socket_tx) = socket.into_split();

    // Establish a Remoc connection with default configuration over the TCP
    // connection and obtain the sending half of the base channel.
    //
    // The connection is always bidirectional, but this end only sends, so the
    // unneeded receiver is dropped.
    let (conn, tx, _rx): (_, _, rch::base::Receiver<()>) =
        remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).await.unwrap();

    // The connection dispatcher must be spawned; it drives everything that
    // travels over this TCP connection.
    tokio::spawn(conn);

    count(tx, 10).await;
}

/// Asks the server to count up to `up_to` and prints each number as it arrives.
async fn count(mut tx: rch::base::Sender<CountReq>, up_to: u32) {
    // Create a new channel. Nothing has been sent yet, so the server does not
    // know about it and no connection has been made for it.
    let (seq_tx, mut seq_rx) = rch::mpsc::channel();

    // Sending the sender half connects the channel to the server, inside the
    // TCP connection that already exists. There is no port to open, nothing to
    // register and no acknowledgement to wait for.
    println!("Asking the server to count up to {up_to}");
    tx.send(CountReq { up_to, seq_tx }).await.unwrap();

    // Receive each number as the server counts it.
    while let Some(i) = seq_rx.recv().await.unwrap() {
        println!("Server counts {i}");
    }

    // recv() returned None, so the server dropped its sender: the sequence is
    // over. A channel closing is how a remote endpoint says it is finished.
    println!("Server is done counting.");
}

Browse it on GitHub