# 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 as `plaza = "0"`.

## Set up the project

Start a new binary crate.

```sh
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:

```toml title="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:

| Parameter | Here | What it is |
|---|---|---|
| `StateType` | `Counter` | The shared state. It needs `Debug + Send + Sync + 'static`. It needs `Clone` only if you call `query_state`. |
| `Op` | `CounterOp` | The operations, in both directions. It needs `Clone + Debug + Send + Sync + 'static`, plus serde to cross a network. |
| `ID` | `UserId` | Who 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:

```rust title="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:

```shell-session
$ 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`:

```rust title="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:

```rust title="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:

```shell-session
$ 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(())`:

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

Run it:

```shell-session
$ 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:

```rust title="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:

```rust title="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:

```rust title="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:

```shell-session
$ 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:

```rust title="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`:

```rust title="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:

```rust title="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:

```rust title="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:

```shell-session
$ 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:

```rust title="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](https://github.com/excsn/plaza/blob/main/docs/guide/01-one-loop-one-truth.md) in the plaza guide explains why the controller is shaped this way. [What each player sees](https://github.com/excsn/plaza/blob/main/docs/guide/10-what-each-player-sees.md) covers snapshots and per-player views. The [shared_counter example](https://github.com/excsn/plaza/tree/main/examples/shared_counter) is a fuller version of this program.

The [plaza usage guide](https://github.com/excsn/plaza/blob/main/core/README.USAGE.md) on GitHub covers guards, uniform snapshots and the rest of the controller. The full API reference is on [docs.rs](https://docs.rs/plaza).