Example: a timer daemon and CLI

A Rust daemon schedules reminders independently of the clients that add them. The same RPC interface works over Unix domain sockets on Unix and local named pipes on Windows. Clients can list pending timers, wait for one completion or watch all future completions.

RPC methods

  • add(NewTimer) schedules a reminder and returns its identifier without waiting for it to finish.
  • list() returns the reminders that have not finished yet.
  • subscribe() returns a broadcast receiver for future completions.

Each client has its own Remoc connection, but the daemon serves the same timer state to all of them. A timer belongs to the daemon, not to the connection that scheduled it.

This is a foreground example with in-memory state, not an installed system service. Stopping the daemon discards its timers. For a smaller introduction to RPC, start with the counter example.

Running the daemon and CLI

Clone the repository:

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

In the first terminal, start the daemon:

cargo run --manifest-path examples/daemon/Cargo.toml --bin timerd

In a second terminal, watch completions:

cargo run --manifest-path examples/daemon/Cargo.toml --bin timerctl -- watch

In a third terminal, add a reminder and list pending timers:

cargo run --manifest-path examples/daemon/Cargo.toml --bin timerctl -- add 30s "Check tea"
cargo run --manifest-path examples/daemon/Cargo.toml --bin timerctl -- list

The add command prints an identifier and exits. When the timer finishes, the watcher prints, for example: 

Finished 1: Check tea

Open another watcher to receive the same notifications in two clients. Completed timers disappear from list.

To wait for one specific timer, use --wait:

cargo run --manifest-path examples/daemon/Cargo.toml --bin timerctl -- add 5s "Check tea" --wait

This client stays connected until its timer finishes. Watchers still receive that completion. Disconnecting the waiting client does not cancel the timer. Use Ctrl+C to stop the daemon, a watcher or a waiting client.

Passing a channel inside a request

NewTimer is a Serde struct containing a duration, a message and an optional rch::oneshot::Sender<Timer>. The channel endpoint travels with the other fields; it needs no separate socket.

  1. With --wait, the CLI creates a oneshot channel. It keeps the receiver and passes the sender in the add request.
  2. The RPC returns the timer's identifier. The daemon keeps the sender until the timer finishes, then sends the completed Timer.
  3. The CLI awaits its receiver over the same connection that carried the RPC call.

watch demonstrates the opposite direction: subscribe() returns a channel receiver from the daemon. The daemon keeps the broadcast sender and can notify several clients. Each watcher receives future events only, not a history.

See remote channels for the channel types and server push and callbacks for other notification patterns.

Unix sockets and Windows named pipes

Only src/unix.rs and src/windows.rs differ between platforms. Both expose a local listener and a connect function; the RPC implementation and Remoc connection setup stay the same.

  • Unix: the default endpoint is timerd.sock, relative to the working directory. Run all programs from the same directory. The socket permits access only by its owner.
  • Windows: the default endpoint is \\.\pipe\remoc-timerd. The daemon creates another listening pipe instance for each new client.

Both programs accept --endpoint before the CLI command to select another socket path or pipe name. An existing endpoint is not taken over by a second daemon.

Normal Unix shutdown removes the socket file. After a forced termination, first ensure the old daemon has stopped, then remove its leftover socket. For real applications, place it in an application-owned directory.

The IPC guide also covers connecting to a child process through stdin and stdout.

Source code

One package contains the shared interface, the timerd and timerctl binaries, and the two platform-specific adapters. provide sends each client its RPC handle; consume receives it. Both helpers start the background connection I/O, while server.serve() dispatches the method calls.

timer-daemon-example

A shared RPC interface, two binaries and local transport adapters in one package.

Cargo.toml

Its dependencies.

[package]
name = "timer-daemon-example"
version = "0.1.0"
edition = "2024"
publish = false

[dependencies]
clap = { version = "4", features = ["derive"] }
humantime = "2"
remoc = { version = "0.20" }
serde = { version = "1", features = ["derive"] }
thiserror = "2"
tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }

[dev-dependencies]
tempfile = "3"

src/lib.rs

Defines the timer service, its data types and a request that carries an optional oneshot sender.

//! Shared interface for a local timer daemon and its command-line client.

#![warn(missing_docs)]

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

#[cfg(unix)]
#[path = "unix.rs"]
pub mod local;

#[cfg(windows)]
#[path = "windows.rs"]
pub mod local;

#[cfg(not(any(unix, windows)))]
compile_error!("This example requires Unix domain sockets or Windows named pipes.");

/// A reminder scheduled by the daemon.
///
/// This ordinary serializable data is used both in RPC results and channel messages.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct Timer {
    /// Identifier assigned by the daemon.
    pub id: u64,
    /// Message to display when the timer finishes.
    pub message: String,
}

/// A reminder to schedule, optionally with its own completion channel.
///
/// Remoc channel endpoints implement Serde too: the sender travels inside this
/// request while its receiver stays with the client. No extra socket is needed.
/// A channel endpoint could also be passed directly as an RPC argument instead.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct NewTimer {
    /// Delay before the reminder finishes.
    pub after: Duration,
    /// Message to display when the timer finishes.
    pub message: String,
    /// An optional channel to notify when this timer finishes.
    pub notify: Option<rch::oneshot::Sender<Timer>>,
}

/// Scheduling a timer failed.
///
/// Application errors travel back as RPC results. `Call` also represents failures
/// to perform the call itself, such as a disconnected daemon.
#[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
pub enum AddError {
    /// The delay cannot be represented by the daemon's clock.
    #[error("the requested delay is too long")]
    InvalidDelay,
    /// All timer identifiers have been used.
    #[error("no timer identifiers remain")]
    IdExhausted,
    /// The remote call failed.
    #[error(transparent)]
    Call(#[from] rtc::CallError),
}

/// Timers belong to the daemon, not to the client that scheduled them.
///
/// The macro generates [`TimerServiceClient`] for making remote calls and
/// [`TimerServiceServerShared`] for dispatching them to our implementation.
/// Calling a client method sends its arguments to the daemon and awaits its result.
///
/// `Shared` selects a server backed by `Arc<T>`: callers share one object,
/// which handles its own synchronization.
#[rtc::remote(server(Shared))]
pub trait TimerService {
    /// Schedule a reminder and return its identifier, without waiting for expiry.
    async fn add(&self, timer: NewTimer) -> Result<u64, AddError>;

    /// List reminders that have not finished yet.
    async fn list(&self) -> Result<Vec<Timer>, rtc::CallError>;

    /// Receive future timer completions over the existing connection.
    ///
    /// The result is a live channel endpoint, not a batch of events. Completions
    /// continue arriving on the receiver after this RPC call has returned.
    async fn subscribe(&self) -> Result<rch::broadcast::Receiver<Timer>, rtc::CallError>;
}

src/bin/timerd.rs

Owns the timers, serves all clients and sends completion notifications.

//! A foreground daemon that keeps timers alive independently of its clients.

use clap::Parser;
use remoc::{codec, prelude::*};
use std::{collections::BTreeMap, sync::Arc};
use timer_daemon_example::{AddError, NewTimer, Timer, TimerService, TimerServiceServerShared, local};
use tokio::{
    sync::Mutex,
    time::{Instant, sleep_until},
};

#[derive(Parser)]
#[command(about = "Run a local reminder daemon")]
struct Args {
    /// Unix socket path or Windows named-pipe name.
    #[arg(long, default_value = local::DEFAULT_ENDPOINT)]
    endpoint: String,
}

#[derive(Default)]
struct State {
    next_id: u64,
    pending: BTreeMap<u64, Timer>,
    events: rch::broadcast::Sender<Timer>,
}

#[derive(Default)]
struct Timers {
    // RPC handlers and independently running timer tasks share this state.
    state: Arc<Mutex<State>>,
}

impl TimerService for Timers {
    async fn add(&self, timer: NewTimer) -> Result<u64, AddError> {
        let NewTimer { after, message, notify } = timer;
        let deadline = Instant::now().checked_add(after).ok_or(AddError::InvalidDelay)?;
        let id = {
            let mut state = self.state.lock().await;
            let id = state.next_id.checked_add(1).ok_or(AddError::IdExhausted)?;
            state.next_id = id;
            state.pending.insert(id, Timer { id, message });
            id
        };

        // The RPC returns the id immediately. This task keeps running even if
        // the client disconnects after scheduling the timer.
        let state = self.state.clone();
        tokio::spawn(async move {
            sleep_until(deadline).await;
            let mut state = state.lock().await;
            let timer = state.pending.remove(&id).expect("scheduled timer exists");
            println!("Finished {}: {}", timer.id, timer.message);

            // This sender arrived inside NewTimer. Its receiver is still in the
            // waiting client, reached over the connection that carried add().
            if let Some(notify) = notify
                && let Err(err) = notify.send(timer.clone())
            {
                eprintln!("Timer notification failed: {err}");
            }

            // Also notify every watcher. Each receiver uses its own client's
            // existing connection; broadcasting needs no new connections.
            match state.events.send(timer) {
                Ok(_) => (),
                // Timers also run when nobody is watching.
                Err(rch::broadcast::SendError::Closed(_)) => (),
                Err(err) => eprintln!("Timer notification failed: {err}"),
            }
        });

        Ok(id)
    }

    async fn list(&self) -> Result<Vec<Timer>, rtc::CallError> {
        Ok(self.state.lock().await.pending.values().cloned().collect())
    }

    async fn subscribe(&self) -> Result<rch::broadcast::Receiver<Timer>, rtc::CallError> {
        Ok(self.state.lock().await.events.subscribe(16))
    }
}

async fn serve(stream: local::Stream, timers: Arc<Timers>) -> Result<(), Box<dyn std::error::Error>> {
    let (reader, writer) = tokio::io::split(stream);
    // Each connection gets an RPC server calling the same shared Timers object.
    // new() also creates a transferable client handle; the object stays here.
    let (server, client) = TimerServiceServerShared::<_, codec::Default>::new(timers);
    // provide() establishes Remoc over the stream, sends the client handle as
    // the initial value, and starts the connection's background I/O task.
    // It returns after sending that value, not when the connection ends.
    remoc::Connect::io(remoc::Cfg::default(), reader, writer).provide(client).await?;
    // Transporting bytes and handling RPC calls are separate: serve() receives
    // method calls and invokes our TimerService implementation.
    server.serve().await?;
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::parse();
    let mut listener = local::Listener::bind(&args.endpoint)?;
    let timers = Arc::new(Timers::default());
    println!("Listening on {}. Press Ctrl+C to stop.", args.endpoint);

    loop {
        tokio::select! {
            stream = listener.accept() => {
                let stream = stream?;
                let timers = timers.clone();
                tokio::spawn(async move {
                    if let Err(err) = serve(stream, timers).await {
                        eprintln!("Client connection failed: {err}");
                    }
                });
            }
            result = tokio::signal::ctrl_c() => {
                result?;
                break;
            }
        }
    }

    Ok(())
}

src/bin/timerctl.rs

Adds and lists timers, waits on a oneshot receiver or watches a returned broadcast receiver.

//! A command-line client for the timer daemon.

use clap::{Parser, Subcommand};
use remoc::prelude::*;
use std::{io::Write, time::Duration};
use timer_daemon_example::{NewTimer, TimerService, TimerServiceClient, local};

#[derive(Parser)]
#[command(about = "Schedule and watch reminders in the local timer daemon")]
struct Args {
    /// Unix socket path or Windows named-pipe name.
    #[arg(long, default_value = local::DEFAULT_ENDPOINT)]
    endpoint: String,
    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Schedule a reminder, for example: add 30s "Check tea".
    Add {
        /// Delay such as 30s, 5m or 1h.
        #[arg(value_parser = humantime::parse_duration)]
        after: Duration,
        /// Reminder message.
        message: String,
        /// Wait for this timer to finish.
        #[arg(long)]
        wait: bool,
    },
    /// List pending reminders.
    List,
    /// Print future timer completions until Ctrl+C.
    Watch,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::parse();
    let stream = local::connect(&args.endpoint).await?;
    let (reader, writer) = tokio::io::split(stream);
    // consume() establishes Remoc and receives the initial value sent by the
    // daemon's provide(): our typed RPC client. It also starts the connection's
    // background I/O task, carrying both RPC traffic and transferred channels.
    let client: TimerServiceClient = remoc::Connect::io(remoc::Cfg::default(), reader, writer).consume().await?;

    match args.command {
        Command::Add { after, message, wait } => {
            let (notify, completion) = if wait {
                // Keep the receiver here and send its sender to the daemon.
                // Unlike a Tokio oneshot, this Remoc endpoint can cross the
                // connection, including as a field in the NewTimer struct.
                let (sender, receiver) = rch::oneshot::channel();
                (Some(sender), Some(receiver))
            } else {
                (None, None)
            };
            // This looks like an ordinary trait call, but sends the request to
            // timerd. The RPC returns the id, not the eventual completion.
            let id = client.add(NewTimer { after, message, notify }).await?;
            println!("Added timer {id}");
            if let Some(completion) = completion {
                // The RPC is finished. Await a separate notification through
                // our oneshot receiver, still using the same connection.
                std::io::stdout().flush()?;
                tokio::select! {
                    timer = completion => {
                        let timer = timer?;
                        println!("Finished {}: {}", timer.id, timer.message);
                    }
                    result = tokio::signal::ctrl_c() => {
                        result?;
                    }
                }
            }
        }
        Command::List => {
            let timers = client.list().await?;
            if timers.is_empty() {
                println!("No pending timers.");
            }
            for timer in timers {
                println!("{}: {}", timer.id, timer.message);
            }
        }
        Command::Watch => {
            // Here the direction is reversed: the daemon returns a receiver
            // as the RPC result and keeps the broadcast sender.
            let mut events = client.subscribe().await?;
            println!("Watching timer completions. Press Ctrl+C to stop.");
            std::io::stdout().flush()?;
            loop {
                tokio::select! {
                    event = events.recv() => {
                        let timer = event?;
                        println!("Finished {}: {}", timer.id, timer.message);
                    }
                    result = tokio::signal::ctrl_c() => {
                        result?;
                        break;
                    }
                }
            }
        }
    }

    Ok(())
}

src/unix.rs

Connects local clients over a Unix domain socket.

//! Local connections over Unix domain sockets.

use std::{fs, io, os::unix::fs::PermissionsExt};
use tokio::net::{UnixListener, UnixStream};

/// Default socket path, relative to the working directory.
pub const DEFAULT_ENDPOINT: &str = "timerd.sock";

/// A connected local client or server.
pub type Stream = UnixStream;

/// A listening socket, removed when the daemon shuts down normally.
pub struct Listener {
    inner: UnixListener,
    endpoint: String,
}

impl Listener {
    /// Bind without removing an existing socket.
    pub fn bind(endpoint: &str) -> io::Result<Self> {
        let listener = Self { inner: UnixListener::bind(endpoint)?, endpoint: endpoint.to_owned() };
        fs::set_permissions(endpoint, fs::Permissions::from_mode(0o600))?;
        Ok(listener)
    }

    /// Accept another client.
    pub async fn accept(&mut self) -> io::Result<Stream> {
        let (stream, _) = self.inner.accept().await?;
        Ok(stream)
    }
}

impl Drop for Listener {
    fn drop(&mut self) {
        match fs::remove_file(&self.endpoint) {
            Ok(()) => (),
            Err(err) if err.kind() == io::ErrorKind::NotFound => (),
            Err(err) => eprintln!("Cannot remove socket {}: {err}", self.endpoint),
        }
    }
}

/// Connect to a running daemon.
pub async fn connect(endpoint: &str) -> io::Result<Stream> {
    UnixStream::connect(endpoint).await
}

src/windows.rs

Connects local clients over instances of a Windows named pipe.

//! Local connections over Windows named pipes.

use std::{io, time::Duration};
use tokio::{
    net::windows::named_pipe::{ClientOptions, NamedPipeClient, NamedPipeServer, ServerOptions},
    time::sleep,
};

/// Default local pipe name.
pub const DEFAULT_ENDPOINT: &str = r"\\.\pipe\remoc-timerd";

/// A connected server-side pipe.
pub type Stream = NamedPipeServer;

/// A named-pipe listener with an instance ready for the next client.
pub struct Listener {
    pipe: NamedPipeServer,
    endpoint: String,
}

impl Listener {
    /// Claim the pipe name, failing if a daemon already owns an instance.
    pub fn bind(endpoint: &str) -> io::Result<Self> {
        let pipe = ServerOptions::new().first_pipe_instance(true).create(endpoint)?;
        Ok(Self { pipe, endpoint: endpoint.to_owned() })
    }

    /// Accept a client and immediately make another instance available.
    pub async fn accept(&mut self) -> io::Result<Stream> {
        self.pipe.connect().await?;
        let next = ServerOptions::new().create(&self.endpoint)?;
        Ok(std::mem::replace(&mut self.pipe, next))
    }
}

/// Connect to a running daemon, retrying when its pipe instances are busy.
pub async fn connect(endpoint: &str) -> io::Result<NamedPipeClient> {
    const ERROR_PIPE_BUSY: i32 = 231;

    loop {
        match ClientOptions::new().open(endpoint) {
            Ok(pipe) => return Ok(pipe),
            Err(err) if err.raw_os_error() == Some(ERROR_PIPE_BUSY) => {
                sleep(Duration::from_millis(50)).await;
            }
            Err(err) => return Err(err),
        }
    }
}

Browse it on GitHub