Share a counter between two clients

For Rust developers who know Tokio and want to see the smallest complete plaza app before building a game or a collaborative tool on it

You will build a counter that two clients share. Each client joins, receives a snapshot of the current value and sends increments. Every change is broadcast back to both. Along the way you will see the types plaza is generic over, why only the controller ever changes the state, how to feed it time, how to ask it a question while it runs and how to shut it down so a closing message still goes out.

Before you start

You need Rust with Cargo. plaza comes from crates.io:

plaza = "0"

Set up the project#

Start a new binary crate.

cargo new counter
cd counter

Add plaza, Tokio and async-trait. The traits you implement are async, so you need async-trait to write them. Your [dependencies] section should look like this:

Cargo.toml
[dependencies]
plaza = "0"
tokio = { version = "1", features = ["macros", "rt"] }
async-trait = "0.1"

Everything here runs in one process over InProcessSession, plaza's in-memory transport, so there are no sockets to set up. The programs use Tokio's single-threaded current_thread runtime and every client reads its inbox in a fixed order, so the output is the same on every run. The ops never cross a network, so they need no serde derives either. A real transport would need them.

Write the rules#

A plaza app is three types of your own and the rules that connect them. The state is the data being shared, here a struct holding one number. The op enum is everything a client can send and everything the server sends back. The id is whatever names a client; a plain u32 works.

Those three are the type parameters nearly everything in plaza is generic over:

ParameterHereWhat it is
StateTypeCounterThe shared state. It needs Debug + Send + Sync + 'static. It needs Clone only if you call query_state.
OpCounterOpThe operations, in both directions. It needs Clone + Debug + Send + Sync + 'static, plus serde to cross a network.
IDUserIdWho acted. Any Clone + Debug + Eq + Hash + Send + Sync + 'static type is an AgentId.

The traits name them in different orders: StateLogic<Op, ID, StateType> and SnapshotProvider<ID, StateType, Op>. The compiler will tell you when you swap two.

The rules are a StateLogic, which has one method. process_input gets the state as &mut and one LogicInput at a time. It returns a LogicOutput saying which ops to send to whom. This is the only place the state is ever changed. The StateController owns the state and calls your rules from its own task, one input after another, so there is nothing to lock.

The snapshot provider decides what a client is sent when it joins. The controller calls create_snapshot once per recipient and passes the agent it is for, so each client could get its own view. Here everyone may see the whole counter, so it ignores the target and returns the value. A snapshot is just another op variant.

Replace src/main.rs with this:

src/main.rs
use std::sync::Arc;

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    controller::StateControllerBuilder,
    session::{InProcessSession, SessionMessage, TargetedOp},
    snapshot::{SnapshotContext, SnapshotError, SnapshotProvider},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
};

type UserId = u32;

#[derive(Debug, Default)]
struct Counter {
    value: i64,
}

#[derive(Clone, Debug)]
enum CounterOp {
    Increment(i64),
    Changed { by: UserId, value: i64 },
    Snapshot { value: i64 },
}

struct CounterRules;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterRules {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        let LogicInput::AgentOps { source, ops } = input else {
            return Ok(LogicOutput::none());
        };
        let Some(by) = source.id_cloned() else {
            return Ok(LogicOutput::none());
        };

        let mut out = Vec::new();
        for op in ops {
            if let CounterOp::Increment(n) = op {
                state.value += n;
                out.push(TargetedOp::new_system_all(vec![CounterOp::Changed {
                    by,
                    value: state.value,
                }]));
            }
        }
        Ok(out.into())
    }
}

struct CounterView;

#[async_trait]
impl SnapshotProvider<UserId, Counter, CounterOp> for CounterView {
    async fn create_snapshot(
        &self,
        state: &Counter,
        _target: Option<&Agent<UserId>>,
        _context: Option<SnapshotContext>,
    ) -> Result<Option<CounterOp>, SnapshotError<UserId>> {
        Ok(Some(CounterOp::Snapshot { value: state.value }))
    }
}

fn show(who: &str, msg: SessionMessage<CounterOp, UserId>) {
    for op in msg.ops {
        match op {
            CounterOp::Snapshot { value } => println!("{who}: snapshot, value is {value}"),
            CounterOp::Changed { by, value } => println!("{who}: user {by} made it {value}"),
            CounterOp::Increment(_) => {}
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<CounterOp, UserId>::new();
    let (_tx, controller) = StateControllerBuilder::new(
        Arc::new(CounterRules),
        session.clone(),
        Arc::new(CounterView),
        Counter::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let alice = Agent::new_human(1);
    let (_, alice_inbox) = session.connect(alice.clone()).await?;
    session
        .client_send(alice.clone(), vec![CounterOp::Increment(5)])
        .await;
    show("alice", alice_inbox.recv().await?);
    show("alice", alice_inbox.recv().await?);

    Ok(())
}

StateControllerBuilder::new takes the rules, the session, the snapshot provider and the initial state. build returns the controller and a CommandSender. The sender is the handle everything outside the controller uses to talk to it. Nothing in plaza spawns a task for you, so you spawn controller.run().

session.connect is what a socket connecting would be on a real transport. It returns an inbox for that client and announces the join. The controller answers by sending the joiner a snapshot. client_send is the client sending ops. A client can send as soon as connect returns, because the controller handles a client's join before any op it sends, so the snapshot always comes first. The rules turn each Increment into a Changed broadcast with TargetedOp::new_system_all. recv on an inbox waits for the next SessionMessage, which holds the sender and a list of ops. show prints the ops a client can receive.

The rules ignore an Increment from an agent with no id. Agent is Human(id), Bot(id) or System. Only the first two carry an id.

Run it:

$ cargo run
alice: snapshot, value is 0
alice: user 1 made it 5

Alice joined while the value was 0, then her own increment came back to her as a broadcast. The client never changes its local copy on its own. It waits for the server to say what happened.

When a provider is a plain function of the state and the recipient, as this one is, you can skip the trait. Write the function and wrap it in SnapshotFn:

src/main.rs
fn view(state: &Counter, _target: Option<&Agent<UserId>>) -> Option<CounterOp> {
    Some(CounterOp::Snapshot { value: state.value })
}

Then pass Arc::new(SnapshotFn(view)) to the builder in place of Arc::new(CounterView) and import plaza::snapshot::SnapshotFn. The output is the same. Implement the trait directly when building a view can fail or has to await.

Add a second client#

Bob joins after Alice has already changed the value. His snapshot should show the value as it is now. His increment should reach both of them.

Replace main with this:

src/main.rs
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<CounterOp, UserId>::new();
    let (_tx, controller) = StateControllerBuilder::new(
        Arc::new(CounterRules),
        session.clone(),
        Arc::new(CounterView),
        Counter::default(),
    )
    .build();
    tokio::spawn(controller.run());

    let alice = Agent::new_human(1);
    let (_, alice_inbox) = session.connect(alice.clone()).await?;
    session
        .client_send(alice.clone(), vec![CounterOp::Increment(5)])
        .await;
    show("alice", alice_inbox.recv().await?);
    show("alice", alice_inbox.recv().await?);

    let bob = Agent::new_human(2);
    let (_, bob_inbox) = session.connect(bob.clone()).await?;
    session
        .client_send(bob.clone(), vec![CounterOp::Increment(3)])
        .await;
    show("bob", bob_inbox.recv().await?);
    show("bob", bob_inbox.recv().await?);
    show("alice", alice_inbox.recv().await?);

    Ok(())
}

Run it:

$ cargo run
alice: snapshot, value is 0
alice: user 1 made it 5
bob: snapshot, value is 5
bob: user 2 made it 8
alice: user 2 made it 8

Bob's snapshot says 5 because the controller built it from the state when he joined. His increment was broadcast to everyone connected, so Alice and Bob both see 8.

The program reads Alice's two messages before Bob connects and reads the inboxes in a fixed order. That only keeps the printed output the same on every run.

Ask the controller a question#

Code outside the controller cannot read the state directly, since the controller owns it. You ask instead. query_with sends a closure to the controller, which runs it on its own task with the state borrowed and sends back the result.

Add query_with to the plaza::controller import, then rename _tx to tx and add this to the end of main, before Ok(()):

src/main.rs
let value = query_with(&tx, |counter| counter.value).await?;
println!("controller: value is {value}");

Run it:

$ cargo run
alice: snapshot, value is 0
alice: user 1 made it 5
bob: snapshot, value is 5
bob: user 2 made it 8
alice: user 2 made it 8
controller: value is 8

Only the i64 is copied. The closure runs inside the controller loop, so it must not block or await. query_state(&tx) returns a copy of the whole state instead. It is the only call that needs Counter to be Clone. On a large state, ask for the field you need.

Feed it time with a TickDriver#

The controller does not keep time. Time reaches the rules the same way ops do, as an input: LogicInput::TimeStep { delta_time }. A TickDriver is a loop that sends those steps at a rate you pick.

The counter will keep track of how long it has been running. Add use std::time::Duration; and tick_driver::TickDriver to the imports, then give the state an uptime field:

src/main.rs
#[derive(Debug, Default)]
struct Counter {
    value: i64,
    uptime: Duration,
}

The rules now match on the input, so they can handle TimeStep as well as ops:

src/main.rs
struct CounterRules;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterRules {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentOps { source, ops } => {
                let Some(by) = source.id_cloned() else {
                    return Ok(LogicOutput::none());
                };
                let mut out = Vec::new();
                for op in ops {
                    if let CounterOp::Increment(n) = op {
                        state.value += n;
                        out.push(TargetedOp::new_system_all(vec![CounterOp::Changed {
                            by,
                            value: state.value,
                        }]));
                    }
                }
                Ok(out.into())
            }
            LogicInput::TimeStep { delta_time } => {
                state.uptime += delta_time;
                Ok(LogicOutput::none())
            }
            _ => Ok(LogicOutput::none()),
        }
    }
}

Add this to main after the query:

src/main.rs
TickDriver::from_hz(100)
    .run_fixed_for(tx.clone(), Duration::from_millis(10), 50)
    .await;
let uptime = query_with(&tx, |counter| counter.uptime).await?;
println!("controller: uptime is {uptime:?}");

Run it:

$ cargo run
alice: snapshot, value is 0
alice: user 1 made it 5
bob: snapshot, value is 5
bob: user 2 made it 8
alice: user 2 made it 8
controller: value is 8
controller: uptime is 500ms

run_fixed_for wakes at 100 Hz and hands the controller exactly 50 steps of exactly 10 ms each, so the uptime is always 500 ms however busy the machine is. The bounded form returns once it has sent them, which suits a demo. A server spawns the unbounded forms and leaves them running:

  • tokio::spawn(TickDriver::from_hz(60).run(tx.clone())) passes the measured time since the last tick. Use it for logic that only integrates, such as a decay or a cooldown.
  • tokio::spawn(TickDriver::from_hz(60).run_fixed(tx.clone(), Duration::from_millis(16))) always passes the same step. Use it whenever anything predicts, replays or rolls back the logic, since a client can only reproduce a simulation stepped in fixed amounts.

Shut it down#

run returns when the controller receives ControllerCommand::Shutdown. It hands back the final state for you to save. Commands already queued when Shutdown arrives are processed first, so a closing message submitted just before it is guaranteed to go out.

The closing message comes from the server itself, which submits it as a system op. Add a variant to the op enum:

src/main.rs
#[derive(Clone, Debug)]
enum CounterOp {
    Increment(i64),
    Changed { by: UserId, value: i64 },
    Snapshot { value: i64 },
    Closing,
}

Then update the rules so an Increment still needs an id while Closing is only accepted from System:

src/main.rs
struct CounterRules;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterRules {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentOps { source, ops } => {
                let mut out = Vec::new();
                for op in ops {
                    match op {
                        CounterOp::Increment(n) => {
                            let Some(by) = source.id_cloned() else {
                                continue;
                            };
                            state.value += n;
                            out.push(TargetedOp::new_system_all(vec![CounterOp::Changed {
                                by,
                                value: state.value,
                            }]));
                        }
                        CounterOp::Closing if source.is_system() => {
                            out.push(TargetedOp::new_system_all(vec![CounterOp::Closing]));
                        }
                        _ => {}
                    }
                }
                Ok(out.into())
            }
            LogicInput::TimeStep { delta_time } => {
                state.uptime += delta_time;
                Ok(LogicOutput::none())
            }
            _ => Ok(LogicOutput::none()),
        }
    }
}

The show function needs an arm for the new variant:

src/main.rs
fn show(who: &str, msg: SessionMessage<CounterOp, UserId>) {
    for op in msg.ops {
        match op {
            CounterOp::Snapshot { value } => println!("{who}: snapshot, value is {value}"),
            CounterOp::Changed { by, value } => println!("{who}: user {by} made it {value}"),
            CounterOp::Closing => println!("{who}: the server is closing"),
            CounterOp::Increment(_) => {}
        }
    }
}

In main, keep the controller's join handle with let running = tokio::spawn(controller.run());, add ControllerCommand to the plaza::controller import and finish with this:

src/main.rs
tx.send(ControllerCommand::SubmitSystemOps {
    source_description: "main".into(),
    ops: vec![CounterOp::Closing],
})
.await?;
tx.send(ControllerCommand::Shutdown).await?;
let last = running.await??;

show("alice", alice_inbox.recv().await?);
show("bob", bob_inbox.recv().await?);
println!("controller: stopped with {last:?}");

Run it:

$ cargo run
alice: snapshot, value is 0
alice: user 1 made it 5
bob: snapshot, value is 5
bob: user 2 made it 8
alice: user 2 made it 8
controller: value is 8
controller: uptime is 500ms
alice: the server is closing
bob: the server is closing
controller: stopped with Counter { value: 8, uptime: 500ms }

SubmitSystemOps reaches the rules as AgentOps from Agent::System, which is how the rules tell it apart from a client trying to send Closing. The broadcast goes out, then Shutdown stops the loop and running.await?? hands back the final Counter. The first ? is Tokio's join error and the second is the controller's own PlazaError.

Here is the whole program:

src/main.rs
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    controller::{ControllerCommand, StateControllerBuilder, query_with},
    session::{InProcessSession, SessionMessage, TargetedOp},
    snapshot::{SnapshotContext, SnapshotError, SnapshotProvider},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
    tick_driver::TickDriver,
};

type UserId = u32;

#[derive(Debug, Default)]
struct Counter {
    value: i64,
    uptime: Duration,
}

#[derive(Clone, Debug)]
enum CounterOp {
    Increment(i64),
    Changed { by: UserId, value: i64 },
    Snapshot { value: i64 },
    Closing,
}

struct CounterRules;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterRules {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentOps { source, ops } => {
                let mut out = Vec::new();
                for op in ops {
                    match op {
                        CounterOp::Increment(n) => {
                            let Some(by) = source.id_cloned() else {
                                continue;
                            };
                            state.value += n;
                            out.push(TargetedOp::new_system_all(vec![CounterOp::Changed {
                                by,
                                value: state.value,
                            }]));
                        }
                        CounterOp::Closing if source.is_system() => {
                            out.push(TargetedOp::new_system_all(vec![CounterOp::Closing]));
                        }
                        _ => {}
                    }
                }
                Ok(out.into())
            }
            LogicInput::TimeStep { delta_time } => {
                state.uptime += delta_time;
                Ok(LogicOutput::none())
            }
            _ => Ok(LogicOutput::none()),
        }
    }
}

struct CounterView;

#[async_trait]
impl SnapshotProvider<UserId, Counter, CounterOp> for CounterView {
    async fn create_snapshot(
        &self,
        state: &Counter,
        _target: Option<&Agent<UserId>>,
        _context: Option<SnapshotContext>,
    ) -> Result<Option<CounterOp>, SnapshotError<UserId>> {
        Ok(Some(CounterOp::Snapshot { value: state.value }))
    }
}

fn show(who: &str, msg: SessionMessage<CounterOp, UserId>) {
    for op in msg.ops {
        match op {
            CounterOp::Snapshot { value } => println!("{who}: snapshot, value is {value}"),
            CounterOp::Changed { by, value } => println!("{who}: user {by} made it {value}"),
            CounterOp::Closing => println!("{who}: the server is closing"),
            CounterOp::Increment(_) => {}
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let session = InProcessSession::<CounterOp, UserId>::new();
    let (tx, controller) = StateControllerBuilder::new(
        Arc::new(CounterRules),
        session.clone(),
        Arc::new(CounterView),
        Counter::default(),
    )
    .build();
    let running = tokio::spawn(controller.run());

    let alice = Agent::new_human(1);
    let (_, alice_inbox) = session.connect(alice.clone()).await?;
    session
        .client_send(alice.clone(), vec![CounterOp::Increment(5)])
        .await;
    show("alice", alice_inbox.recv().await?);
    show("alice", alice_inbox.recv().await?);

    let bob = Agent::new_human(2);
    let (_, bob_inbox) = session.connect(bob.clone()).await?;
    session
        .client_send(bob.clone(), vec![CounterOp::Increment(3)])
        .await;
    show("bob", bob_inbox.recv().await?);
    show("bob", bob_inbox.recv().await?);
    show("alice", alice_inbox.recv().await?);

    let value = query_with(&tx, |counter| counter.value).await?;
    println!("controller: value is {value}");

    TickDriver::from_hz(100)
        .run_fixed_for(tx.clone(), Duration::from_millis(10), 50)
        .await;
    let uptime = query_with(&tx, |counter| counter.uptime).await?;
    println!("controller: uptime is {uptime:?}");

    tx.send(ControllerCommand::SubmitSystemOps {
        source_description: "main".into(),
        ops: vec![CounterOp::Closing],
    })
    .await?;
    tx.send(ControllerCommand::Shutdown).await?;
    let last = running.await??;

    show("alice", alice_inbox.recv().await?);
    show("bob", bob_inbox.recv().await?);
    println!("controller: stopped with {last:?}");

    Ok(())
}

Where to go next#

The rules here ignore AgentJoined and AgentLeft. A real app usually seats a player on join and frees the seat on leave. The session reports both the same way it reported the join here. InProcessSession::disconnect is the in-process equivalent of a socket closing. When a change alters what clients should see, the rules can push fresh views without waiting for a join by returning LogicOutput::ops(ops).and_snapshot(SnapshotRequest::to(recipients)).

One loop, one truth in the plaza guide explains why the controller is shaped this way. What each player sees covers snapshots and per-player views. The shared_counter example is a fuller version of this program.

The plaza usage guide on GitHub covers guards, uniform snapshots and the rest of the controller. The full API reference is on docs.rs.