A counter over remote trait calls: a Remoc example

The #[rtc::remote] attribute generates a client that implements the trait and server types that dispatch calls to the trait implementation. Calling a method on the client runs it on the server and returns its result.

The watch() method returns a watch receiver and count_to_value() an mpsc receiver. Both receivers continue to deliver values after the method call returns.

Running it

Clone the repository:

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

Start the server:

cargo run --manifest-path examples/rtc/Cargo.toml -p counter-server

Then, in another terminal, start the client:

cargo run --manifest-path examples/rtc/Cargo.toml -p counter-client

The program

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

counter

Defines the remote trait shared by the client and server.

`counter/Cargo.toml`

Its dependencies.

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

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

`counter/src/lib.rs`

Defines the trait carrying #[rtc::remote] and the error type returned by its methods. Two methods return channel receivers.

//! This library crate defines the remote counting service.
//!
//! The client and server depend on it.
#![warn(missing_docs)]

use remoc::prelude::*;
use std::time::Duration;

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

/// Increasing the counter failed.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum IncreaseError {
    /// An overflow would occur.
    Overflow {
        /// The current value of the counter.
        current_value: u32,
    },
    /// The RTC call failed.
    Call(rtc::CallError),
}

impl From<rtc::CallError> for IncreaseError {
    fn from(err: rtc::CallError) -> Self {
        Self::Call(err)
    }
}

/// Remote counting service.
#[rtc::remote(server(SharedMut))]
pub trait Counter {
    /// Obtain the current value of the counter.
    async fn value(&self) -> Result<u32, rtc::CallError>;

    /// Watch the current value of the counter for immediate notification
    /// when it changes.
    async fn watch(&mut self) -> Result<rch::watch::Receiver<u32>, rtc::CallError>;

    /// Increase the counter's value by the provided number.
    async fn increase(&mut self, #[doc = "increment value"] by: u32) -> Result<(), IncreaseError>;

    /// Counts to the current value of the counter with the specified
    /// delay between each step.
    async fn count_to_value(
        &self, step: u32, delay: Duration,
    ) -> Result<rch::mpsc::Receiver<u32>, rtc::CallError>;
}

counter-server

Holds the counter and serves it over each accepted connection.

`counter-server/Cargo.toml`

Its dependencies.

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

[dependencies]
counter = { path = "../counter" }
remoc = { version = "0.20" }
tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "time"] }
tracing = "0.1"
tracing-subscriber = "0.3.22"

`counter-server/src/main.rs`

Implements the trait on the counter state and runs a generated server for each connection.

//! This crate implements the server of the remote counting service.
#![warn(missing_docs)]

use remoc::{codec, prelude::*};
use std::{net::Ipv4Addr, sync::Arc, time::Duration};
use tokio::{net::TcpListener, sync::RwLock, time::sleep};
use tracing::{Instrument, info_span};

use counter::{Counter, CounterServerSharedMut, IncreaseError, TCP_PORT};

/// Server object for the counting service, keeping the state.
#[derive(Default)]
pub struct CounterObj {
    /// The current value.
    value: u32,
    /// The subscribed watchers.
    watchers: Vec<rch::watch::Sender<u32>>,
}

/// Implementation of remote counting service.
impl Counter for CounterObj {
    async fn value(&self) -> Result<u32, rtc::CallError> {
        Ok(self.value)
    }

    async fn watch(&mut self) -> Result<rch::watch::Receiver<u32>, rtc::CallError> {
        // Create watch channel.
        let (tx, rx) = rch::watch::channel(self.value);
        // Keep the sender half in the watchers vector.
        self.watchers.push(tx);
        // And return the receiver half.
        Ok(rx)
    }

    async fn increase(&mut self, by: u32) -> Result<(), IncreaseError> {
        tracing::info!(%by, "increase");

        // Perform the addition if it does not overflow the counter.
        match self.value.checked_add(by) {
            Some(new_value) => self.value = new_value,
            None => return Err(IncreaseError::Overflow { current_value: self.value }),
        }

        // Notify all watchers and keep only the ones that are not disconnected.
        let value = self.value;
        self.watchers.retain(|watch| !watch.send(value).into_disconnected().unwrap());

        Ok(())
    }

    async fn count_to_value(
        &self, step: u32, delay: Duration,
    ) -> Result<rch::mpsc::Receiver<u32>, rtc::CallError> {
        // Create mpsc channel for counting.
        let (tx, rx) = rch::mpsc::channel();

        // Spawn a task to perform the counting.
        let value = self.value;
        tokio::spawn(async move {
            // Counting loop.
            for i in (0..value).step_by(step as usize) {
                // Send the value.
                if tx.send(i).await.into_disconnected().unwrap() {
                    // Abort the counting if the client dropped the
                    // receive half or disconnected.
                    break;
                }

                // Wait the specified delay.
                sleep(delay).await;
            }
        });

        // Return the receive half of the counting channel.
        Ok(rx)
    }
}

#[tokio::main]
async fn main() {
    // Initialize logging.
    tracing_subscriber::fmt::init();

    // Create a counter object that will be shared between all clients.
    // You could also create one counter object per connection.
    let counter_obj = Arc::new(RwLock::new(CounterObj::default()));

    // Listen to TCP connections using Tokio.
    // In reality you would probably use TLS or WebSockets over HTTPS.
    println!("Listening on port {}. Press Ctrl+C to exit.", TCP_PORT);
    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);

        // Create a new shared reference to the counter object.
        let counter_obj = counter_obj.clone();

        // Spawn a task for each incoming connection.
        tokio::spawn(
            async move {
                tracing::info!("serving");

                // Create a server proxy and client for the accepted connection.
                //
                // The server proxy executes all incoming method calls on the shared counter_obj
                // with a request queue length of 1.
                //
                // Current limitations of the Rust compiler require that we explicitly
                // specify the codec.
                let (server, client) = CounterServerSharedMut::<_, codec::Default>::new(counter_obj);

                // Establish a Remoc connection with default configuration over the TCP connection and
                // provide (i.e. send) the counter client to the client.
                remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).provide(client).await.unwrap();

                // Serve incoming requests from the client on this task.
                // `true` indicates that requests are handled in parallel.
                server.serve().await.unwrap();
            }
            .instrument(info_span!("incoming", %addr)),
        );
    }
}

counter-client

Exercises the counter: reads it, increases it, watches it and streams from it.

`counter-client/Cargo.toml`

Its dependencies.

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

[dependencies]
counter = { path = "../counter" }
remoc = { version = "0.20" }
tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "time"] }
tracing-subscriber = "0.3.22"

`counter-client/src/main.rs`

Uses the generated client to call the trait methods and receive updates.

//! This crate implements the client of the remote counting service.
#![warn(missing_docs)]

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

use counter::{Counter, CounterClient, TCP_PORT};

#[tokio::main]
async fn main() {
    // Initialize logging.
    tracing_subscriber::fmt::init();

    // 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
    // consume (i.e. receive) the counter client from the server.
    let mut client: CounterClient =
        remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).consume().await.unwrap();

    // Subscribe to the counter watch and print each value change.
    println!("Subscribing to counter change notifications");
    let mut watch_rx = client.watch().await.unwrap();
    let watch_task = tokio::spawn(async move {
        while watch_rx.changed().await.is_ok() {
            let value = watch_rx.borrow_and_update().unwrap();
            println!("Counter change notification: {}", *value);
        }
    });
    println!("Done!");

    // Print current value.
    let value = client.value().await.unwrap();
    println!("Current counter value is {}\n", value);

    // Increase counter value.
    println!("Increasing counter value by 5");
    client.increase(5).await.unwrap();
    println!("Done!\n");

    // Print new value.
    let value = client.value().await.unwrap();
    println!("New counter value is {}\n", value);

    // Let the server count to the current value.
    println!("Asking the server to count to the current value with a step delay of 300ms...");
    let mut rx = client.count_to_value(1, Duration::from_millis(300)).await.unwrap();
    while let Ok(Some(i)) = rx.recv().await {
        println!("Server counts {}", i);
    }
    println!("Server is done counting.\n");

    // Wait for watch task.
    println!("Server exercise is done.");
    println!("You can now press Ctrl+C to exit or continue watching for value ");
    println!("change notification caused by other clients.");
    watch_task.await.unwrap();
}

Browse it on GitHub