# Put the controller on a real socket

**For:** Rust developers who have a plaza controller running in process and want to serve it to real clients over WebSockets.

You will serve a shared counter over WebSockets on localhost and connect clients to it with `plaza_ws`. Along the way you will read the frames the session puts on the wire, declare a protocol version so a stale client finds out on connect, admit connections by a credential and limit both how fast one connection may send and how much it is sent.

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_net
cd counter_net
```

`plaza` holds the controller and your rules. `plaza_session` puts a real transport under it and `plaza_ws` is the client. `plaza_wire` is the wire vocabulary both ends share: the codec, the frame kinds and the protocol version. Your `Cargo.toml` should end like this:

```toml title="Cargo.toml"
[dependencies]
actix-web = "4"
async-trait = "0.1"
plaza = "0"
plaza_session = "0"
plaza_wire = "0"
plaza_ws = { version = "0", features = ["native", "pump"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

[build-dependencies]
plaza_wire = { version = "0", default-features = false, features = ["build"] }
```

`plaza_session` ships two transports, `ActixWsPlazaSession` for WebSockets on actix-web and `TcpPlazaSession` for length-delimited TCP. Both are thin pumps over the same connection manager, so everything the server does in this tutorial works the same on either. `plaza_ws` is a WebSocket client, so the server here is the actix one. The `native` feature gives `plaza_ws` its desktop backend and `pump` adds `FramePump`, the client half of plaza's protocol. The build dependency is for the protocol version later on.

The server and its clients share one process so you can run the whole thing at once. The server runs on its own thread inside an actix system and listens on `127.0.0.1:5770`. The clients run on the main thread and poll their sockets in a loop, the way a game polls once per frame.

## Write the counter

The rules know nothing about sockets. Put them in `src/counter.rs`:

```rust title="src/counter.rs"
use async_trait::async_trait;
use plaza::agent::Agent;
use plaza::error::{SnapshotError, StateLogicError};
use plaza::snapshot::{SnapshotContext, SnapshotProvider};
use plaza::state_logic::{LogicInput, LogicOutput, SnapshotRequest, StateLogic};
use serde::{Deserialize, Serialize};

pub type UserId = u64;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CounterOp {
    Increment(i64),
    Value(i64),
}

#[derive(Debug, Default)]
pub struct Counter {
    value: i64,
    here: Vec<Agent<UserId>>,
}

pub struct CounterLogic;

#[async_trait]
impl StateLogic<CounterOp, UserId, Counter> for CounterLogic {
    async fn process_input(
        &self,
        state: &mut Counter,
        input: LogicInput<CounterOp, UserId>,
    ) -> Result<LogicOutput<CounterOp, UserId>, StateLogicError> {
        match input {
            LogicInput::AgentJoined { agent } => state.here.push(agent),
            LogicInput::AgentLeft { agent_id } => state.here.retain(|a| a.id() != Some(&agent_id)),
            LogicInput::AgentOps { ops, .. } => {
                for op in ops {
                    if let CounterOp::Increment(by) = op {
                        state.value += by;
                    }
                }
                return Ok(
                    LogicOutput::none().and_snapshot(SnapshotRequest::to(state.here.clone()))
                );
            }
            LogicInput::TimeStep { .. } => {}
        }
        Ok(LogicOutput::none())
    }
}

pub struct CounterSnapshotter;

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

`Counter` keeps the value and a roster of who is connected, because a snapshot request names its recipients. Each batch of ops changes the value and asks for a fresh snapshot for everyone. The snapshot is a `CounterOp::Value`. When a client joins, the controller sends it a snapshot of its own, so a new client learns the current value without asking.

## Serve it over WebSockets

Replace `src/main.rs` with this:

```rust title="src/main.rs"
mod counter;

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_session::ActixWsPlazaSession;
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, UserId};

const ADDR: &str = "127.0.0.1:5770";
const URL: &str = "ws://127.0.0.1:5770/ws";

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

static NEXT_USER: AtomicU64 = AtomicU64::new(1);

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    let id = NEXT_USER.fetch_add(1, Ordering::Relaxed);
    session.handle_connection(&req, stream, Agent::new_human(id))
}

fn serve(ready: mpsc::Sender<()>) {
    actix_web::rt::System::new().block_on(async move {
        let session = CounterSession::new();
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let session = web::Data::new(session);
        let server = HttpServer::new(move || {
            App::new()
                .app_data(session.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(()).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    name: &'static str,
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
}

impl Client {
    fn connect(name: &'static str) -> Self {
        let pump = FramePump::connect(URL, JsonCodec, 0).expect("connect");
        Client {
            name,
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => panic!("{}: closed: {closed}", self.name),
                Arrival::Opened | Arrival::Mismatch { .. } => {}
            }
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                println!("{}: value {value}", self.name);
                return value;
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: no value within 5s", self.name);
    }
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    started.recv().expect("the server started");

    let mut alice = Client::connect("alice");
    alice.next_value();
    alice.send(CounterOp::Increment(5));
    alice.next_value();

    let mut bob = Client::connect("bob");
    bob.next_value();
    bob.send(CounterOp::Increment(2));
    bob.next_value();
    alice.next_value();
}
```

`serve` builds the session, hands it to the controller and puts the session in actix's app data so the route can reach it. `ActixWsPlazaSession::new()` spawns the task that decodes inbound frames, so it has to be called inside a Tokio runtime. The actix system's `block_on` is one. The `_commands` binding keeps the controller's command sender alive for as long as the server runs.

The route does one thing. `handle_connection` completes the WebSocket handshake, registers the connection as the `Agent` you give it and runs the socket's pump until it closes. Here each connection gets the next number as its id.

On the client side, `FramePump::connect` opens the socket and returns at once. The `0` is the protocol version this client declares. Zero declares nothing. `poll` never blocks. It sends what is queued, answers the server's latency probes and hands back only the arrivals your code has to act on: `Opened`, `Ops`, `Mismatch` and `Closed`. An `Ops` arrival carries the frame undecoded, so `Client::poll` decodes the body itself with the same codec the server uses.

Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
```

Alice joins and gets the current value as her join snapshot. Each client reads that snapshot before it sends anything, so the lines print in the same order on every run. Her increment comes back to her as a new snapshot. Bob joins at 5, adds 2 and both clients see 7. When `main` returns the process exits and takes the server thread with it.

## Read the frames on the wire

`FramePump` hides the framing. To see it, connect a bare `plaza_ws` socket and print what arrives. Add `use plaza_wire::frame::{self, Kind};` and `use plaza_ws::{Event, Socket};` to the imports in `src/main.rs`, then add these two functions above `main`:

```rust title="src/main.rs"
fn show(direction: &str, bytes: &[u8]) {
    let (tag, body) = frame::split(bytes).expect("a frame has a kind byte");
    match Kind::from_byte(tag) {
        Some(Kind::Ping | Kind::Pong) => {}
        Some(kind) => println!(
            "{direction}: [{tag}] {kind:?} {}",
            String::from_utf8_lossy(body)
        ),
        None => println!("{direction}: [{tag}] unknown kind"),
    }
}

fn peek() {
    let mut socket = plaza_ws::connect(URL).expect("connect");
    let mut events = Vec::new();
    let mut received = 0;
    while received < 2 {
        socket.poll(&mut events);
        for event in events.drain(..) {
            let (direction, bytes) = match event {
                Event::Text(text) => ("text in", text.into_bytes()),
                Event::Message(bytes) => ("binary in", bytes),
                Event::Open | Event::Closed(_) => continue,
            };
            show(direction, &bytes);
            received += 1;
            if received == 1 {
                let mut buf = Vec::new();
                frame::begin(Kind::Ops, &mut buf);
                JsonCodec
                    .encode_into(&vec![CounterOp::Increment(1)], &mut buf)
                    .expect("encode");
                show("binary out", &buf);
                socket.send(&buf).expect("send");
            }
        }
        thread::sleep(Duration::from_millis(5));
    }
}
```

Finally call `peek();` as the last line of `main`. Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
text in: [0] Ops [{"Value":7}]
binary out: [0] Ops [{"Increment":1}]
text in: [0] Ops [{"Value":8}]
```

Every plaza frame is one kind byte followed by the codec's encoding of the body. Nothing else is on the wire. Kind 0 is `Ops` and its body is the array of ops. There is no envelope around it and no sender field: the server knows who sent a frame from the connection it read it on. The other kinds are `Hello`, `Ping`, `Pong`, `Credential` and `Goodbye`. A receiver skips a kind it does not know rather than closing the connection.

The body goes through a `WireCodec`. `JsonCodec` is the default and its `is_text()` is true, so the server sends JSON as WebSocket text frames you could read with `JSON.parse` in a browser or with `websocat`. The server accepts binary frames too, which is how `peek` sent its increment. `MsgPackCodec` behind the `msgpack` feature is the compact choice: pass it to `ActixWsPlazaSession::with_codec` on the server and to `FramePump::connect` on the client. You can also implement `WireCodec` for a format of your own and keep the kind byte in front of it.

`show` skips `Ping` and `Pong`. The session probes every connection to measure its round trip, so a few of those arrive while `peek` waits and how many depends on timing. `FramePump` answers them for you.

`peek` has done its job. Remove it, `show`, the call at the end of `main` and the two imports before you go on, because in the next step the server sends a `Hello` before anything else.

## Declare a protocol version

A client is a separate build from the server. When the shape of `CounterOp` changes, a client built against the old shape still connects and fails one message at a time. Plaza's answer is a protocol version derived from your type definitions at build time and exchanged once per connection in a `Hello` frame.

Create `build.rs` next to `Cargo.toml`:

```rust title="build.rs"
fn main() {
    plaza_wire::build::Wire::detect().emit();
}
```

`Wire::detect()` starts from the types you tag as roots, walks their field types and hashes the definitions it reaches. Comments, formatting and `impl` blocks do not move the number. A field, a variant or a reordering does. Tag `CounterOp` and pull the number in at the top of `src/counter.rs`:

```rust title="src/counter.rs"
pub type UserId = u64;

include!(concat!(env!("OUT_DIR"), "/wire_protocol.rs"));
pub const PROTOCOL: u32 = WIRE_PROTOCOL;

/// plaza-wire: root
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum CounterOp {
```

The `/// plaza-wire: root` line is the tag the build script looks for. Now replace `src/main.rs`:

```rust title="src/main.rs"
mod counter;

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_session::ActixWsPlazaSession;
use plaza_wire::frame::ProtocolVersion;
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, PROTOCOL, UserId};

const ADDR: &str = "127.0.0.1:5770";
const URL: &str = "ws://127.0.0.1:5770/ws";

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

static NEXT_USER: AtomicU64 = AtomicU64::new(1);

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    let id = NEXT_USER.fetch_add(1, Ordering::Relaxed);
    session.handle_connection(&req, stream, Agent::new_human(id))
}

fn serve(ready: mpsc::Sender<Arc<CounterSession>>) {
    actix_web::rt::System::new().block_on(async move {
        let session = CounterSession::with_protocol(JsonCodec, ProtocolVersion(PROTOCOL));
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let data = web::Data::new(session.clone());
        let server = HttpServer::new(move || {
            App::new()
                .app_data(data.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(session).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    name: &'static str,
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
}

impl Client {
    fn connect(name: &'static str, protocol: u32) -> Self {
        let pump = FramePump::connect(URL, JsonCodec, protocol).expect("connect");
        Client {
            name,
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => panic!("{}: closed: {closed}", self.name),
                Arrival::Mismatch { .. } => {
                    println!("{}: the server speaks another wire format", self.name)
                }
                Arrival::Opened => {}
            }
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                println!("{}: value {value}", self.name);
                return value;
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: no value within 5s", self.name);
    }
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    let session = started.recv().expect("the server started");

    let mut alice = Client::connect("alice", PROTOCOL);
    alice.next_value();
    alice.send(CounterOp::Increment(5));
    alice.next_value();

    let mut bob = Client::connect("bob", PROTOCOL);
    bob.next_value();
    bob.send(CounterOp::Increment(2));
    bob.next_value();
    alice.next_value();

    let mut stale = Client::connect("stale", PROTOCOL.wrapping_add(1));
    stale.next_value();
    stale.send(CounterOp::Increment(1));
    stale.next_value();

    for (name, id) in [("alice", 1), ("bob", 2), ("stale", 3)] {
        let same = session.protocol(&id) == Some(ProtocolVersion(PROTOCOL));
        println!("server: {name} speaks our wire format: {same}");
    }
}
```

The session is built with `with_protocol`, so it sends a `Hello` with `PROTOCOL` to every client before anything else. Each client passes its own version to `FramePump::connect` and sends its own `Hello` when the socket opens. `serve` now also hands the session back to `main` once the port is bound, so `main` can ask the server what each client declared.

The stale client stands in for a build from last week. It declares a different number, as an older build of this crate would after a change to `CounterOp`.

Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
stale: the server speaks another wire format
stale: value 7
stale: value 8
server: alice speaks our wire format: true
server: bob speaks our wire format: true
server: stale speaks our wire format: false
```

The stale client hears about the skew as `Arrival::Mismatch` before its first snapshot and the server records what it declared, readable with `session.protocol(&id)`. Neither side refuses the connection and the stale client still gets its values. A version is a build hash, so a peer that only recompiled looks the same as one whose types changed. Plaza leaves the decision to you. You might show a banner, tell a browser page to reload or answer the client with an op of your own. A peer that declares nothing, as the first two programs did with `0`, counts as agreeing.

## Admit connections by credential

`handle_connection` takes the agent from the route, which works when the route can tell who is connecting from a cookie or the URL. A browser or mobile client cannot set a header on its WebSocket upgrade, so plaza lets it present a credential in a frame right after its `Hello`. A `ConnectionAdmitter` on the server reads it and decides. Replace `src/main.rs`:

```rust title="src/main.rs" {9,12,15,21,25,27-39,46,80,84-87,93,115,132-134,138,140-156,165,170,176-177,179-184}
mod counter;

use std::collections::VecDeque;
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use async_trait::async_trait;
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_session::{ActixWsPlazaSession, ConnectionAdmission, ConnectionAdmitter, Farewell, Peer};
use plaza_wire::frame::ProtocolVersion;
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, Closed, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, PROTOCOL, UserId};

const ADDR: &str = "127.0.0.1:5770";
const URL: &str = "ws://127.0.0.1:5770/ws";
const UNKNOWN_TOKEN: u16 = 4403;

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

struct Tokens;

#[async_trait]
impl ConnectionAdmitter<UserId> for Tokens {
    async fn admit(&self, credential: &[u8], _peer: &Peer) -> ConnectionAdmission<UserId> {
        match credential {
            b"alice-token" => ConnectionAdmission::Admitted(Agent::new_human(1)),
            b"bob-token" => ConnectionAdmission::Admitted(Agent::new_human(2)),
            b"carol-token" => ConnectionAdmission::Admitted(Agent::new_human(3)),
            _ => ConnectionAdmission::Refused(
                Farewell::new(UNKNOWN_TOKEN).with_detail("unknown token".as_bytes()),
            ),
        }
    }
}

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    session.admit_connection(&req, stream, Arc::new(Tokens))
}

fn serve(ready: mpsc::Sender<Arc<CounterSession>>) {
    actix_web::rt::System::new().block_on(async move {
        let session = CounterSession::with_protocol(JsonCodec, ProtocolVersion(PROTOCOL));
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let data = web::Data::new(session.clone());
        let server = HttpServer::new(move || {
            App::new()
                .app_data(data.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(session).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    name: &'static str,
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
    closed: Option<Closed>,
}

impl Client {
    fn connect(name: &'static str, token: &str) -> Self {
        let pump = FramePump::connect(URL, JsonCodec, PROTOCOL)
            .expect("connect")
            .credential(token);
        Client {
            name,
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
            closed: None,
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => self.closed = Some(closed),
                Arrival::Mismatch { .. } => {
                    println!("{}: the server speaks another wire format", self.name)
                }
                Arrival::Opened => {}
            }
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                println!("{}: value {value}", self.name);
                return value;
            }
            if let Some(closed) = &self.closed {
                panic!("{}: closed: {closed}", self.name);
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: no value within 5s", self.name);
    }

    fn wait_closed(&mut self) {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(closed) = &self.closed {
                let detail = String::from_utf8_lossy(closed.detail.as_deref().unwrap_or_default());
                println!(
                    "{}: closed with {:?} ({detail}), refused: {}",
                    self.name,
                    closed.code,
                    closed.refused()
                );
                return;
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: still open after 5s", self.name);
    }
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    let session = started.recv().expect("the server started");

    let mut alice = Client::connect("alice", "alice-token");
    alice.next_value();
    alice.send(CounterOp::Increment(5));
    alice.next_value();

    let mut bob = Client::connect("bob", "bob-token");
    bob.next_value();
    bob.send(CounterOp::Increment(2));
    bob.next_value();
    alice.next_value();

    let mut mallory = Client::connect("mallory", "guessed-token");
    mallory.wait_closed();

    let stats = session.stats();
    println!(
        "server: admitted {}, refused {}",
        stats.admitted(),
        stats.refused()
    );
}
```

`Tokens::admit` gets the credential bytes exactly as the client sent them. Plaza verifies nothing, so a token, a signed ticket and an account number are all the same to it. `Admitted` registers the connection as that agent and the join fires from there. `Refused` carries a `Farewell`: the server writes it to the client as a `Goodbye` frame and closes the socket without registering anything. Codes 4000 to 4999 belong to your application.

The route calls `admit_connection` instead of `handle_connection`. It completes the handshake and holds the socket, unregistered, until the admitter answers. On the client, `.credential(token)` makes the pump send the token after its `Hello` and before any op. The connection ids now come from the tokens, so `NEXT_USER` is gone. Carol's token is there for the next step.

Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
mallory: closed with Some(4403) (unknown token), refused: true
server: admitted 2, refused 1
```

Mallory's close arrives as `Arrival::Closed` with the goodbye's code and detail. `refused()` is true for any 4xxx code, which tells a client the server decided and reconnecting with the same credential will be refused again. The session counts both outcomes in its `stats()`. A socket that never presents a credential is closed with 4408 after five seconds. An op sent before the credential closes it with 4401.

## Limit how fast a connection sends

`SessionOptions` holds everything a session is sized and limited by. `rate_limit_inbound` puts a token bucket on each connection's inbound frames. It is judged on that connection's own task before the queue every connection shares, so a flood costs the flooder and nobody else. Change the imports:

```rust title="src/main.rs"
use plaza_session::{
    ActixWsPlazaSession, ConnectionAdmission, ConnectionAdmitter, Farewell, Peer, Rate,
    SessionOptions,
};
```

In `serve`, build the session from options:

```rust title="src/main.rs"
        let options = SessionOptions::with_protocol(ProtocolVersion(PROTOCOL))
            .rate_limit_inbound(Rate::per_second(5.0).burst(5));
        let session = CounterSession::with_options(JsonCodec, options);
```

Then add a client that floods at the end of `main`:

```rust title="src/main.rs"
    let mut carol = Client::connect("carol", "carol-token");
    carol.next_value();
    for _ in 0..20 {
        carol.send(CounterOp::Increment(1));
    }
    for _ in 0..5 {
        carol.next_value();
    }
    thread::sleep(Duration::from_millis(500));
    carol.poll();
    println!("carol: {} more values", carol.values.len());

    let shed = session.manager().agent_inbound(&3).shed;
    println!("server: shed {shed} of carol's frames");
```

Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
mallory: closed with Some(4403) (unknown token), refused: true
server: admitted 2, refused 1
carol: value 7
carol: value 8
carol: value 9
carol: value 10
carol: value 11
carol: value 12
carol: 0 more values
server: shed 15 of carol's frames
```

The bucket holds five frames and refills at five a second. Carol's twenty increments arrive within a few milliseconds, so five are admitted and fifteen are shed. A shed frame is dropped and the connection stays open. Those are ops the client believes arrived, which is fine for an eager client whose next op replaces the last and wrong for a stream where every op counts once. For that case `Rate::per_second(..).disconnecting()` closes the connection instead, with code 1008. There is no default rate, so a session without one admits everything.

## Budget what a connection is sent

The outbound side works differently. The transport never withholds a frame it has been handed. A budget answers one question instead: is this connection owed a frame right now? Your `SnapshotProvider` asks it before building a snapshot for a recipient and returns `Ok(None)` to skip one that is not owed. A client on a slow link then gets fewer whole snapshots instead of a queue that overflows.

The provider needs the connection manager to ask. In `src/counter.rs`, add `use std::sync::Arc;` and `use plaza_session::ConnectionManager;` to the imports and change the snapshotter:

```rust title="src/counter.rs"
pub struct CounterSnapshotter {
    pub manager: Arc<ConnectionManager<UserId>>,
}

#[async_trait]
impl SnapshotProvider<UserId, Counter, CounterOp> for CounterSnapshotter {
    async fn create_snapshot(
        &self,
        state: &Counter,
        target: Option<&Agent<UserId>>,
        _context: Option<SnapshotContext>,
    ) -> Result<Option<CounterOp>, SnapshotError<UserId>> {
        if let Some(id) = target.and_then(|agent| agent.id())
            && !self.manager.agent_owed(id)
        {
            return Ok(None);
        }
        Ok(Some(CounterOp::Value(state.value)))
    }
}
```

Here is the whole of `src/main.rs` with a budget for every connection and a tighter one for a client on a slow link:

```rust title="src/main.rs" {12-15,37,55-58,62-64,195-205,207-208,210-222,224-226,228-229}
mod counter;

use std::collections::VecDeque;
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};

use actix_web::{App, HttpRequest, HttpResponse, HttpServer, web};
use async_trait::async_trait;
use plaza::agent::Agent;
use plaza::controller::StateControllerBuilder;
use plaza_session::{
    ActixWsPlazaSession, ConnectionAdmission, ConnectionAdmitter, Farewell, OutboundBudget, Peer,
    Rate, SessionOptions,
};
use plaza_wire::frame::ProtocolVersion;
use plaza_wire::{JsonCodec, WireCodec};
use plaza_ws::pump::{Arrival, Closed, FramePump};

use counter::{Counter, CounterLogic, CounterOp, CounterSnapshotter, PROTOCOL, UserId};

const ADDR: &str = "127.0.0.1:5770";
const URL: &str = "ws://127.0.0.1:5770/ws";
const UNKNOWN_TOKEN: u16 = 4403;

type CounterSession = ActixWsPlazaSession<CounterOp, UserId>;

struct Tokens;

#[async_trait]
impl ConnectionAdmitter<UserId> for Tokens {
    async fn admit(&self, credential: &[u8], _peer: &Peer) -> ConnectionAdmission<UserId> {
        match credential {
            b"alice-token" => ConnectionAdmission::Admitted(Agent::new_human(1)),
            b"bob-token" => ConnectionAdmission::Admitted(Agent::new_human(2)),
            b"carol-token" => ConnectionAdmission::Admitted(Agent::new_human(3)),
            b"dave-token" => ConnectionAdmission::Admitted(Agent::new_human(4)),
            _ => ConnectionAdmission::Refused(
                Farewell::new(UNKNOWN_TOKEN).with_detail("unknown token".as_bytes()),
            ),
        }
    }
}

async fn ws_route(
    req: HttpRequest,
    stream: web::Payload,
    session: web::Data<Arc<CounterSession>>,
) -> Result<HttpResponse, actix_web::Error> {
    session.admit_connection(&req, stream, Arc::new(Tokens))
}

fn serve(ready: mpsc::Sender<Arc<CounterSession>>) {
    actix_web::rt::System::new().block_on(async move {
        let options = SessionOptions::with_protocol(ProtocolVersion(PROTOCOL))
            .rate_limit_inbound(Rate::per_second(5.0).burst(5))
            .budget_outbound(OutboundBudget::frames_per_second(30.0));
        let session = CounterSession::with_options(JsonCodec, options);
        let (_commands, controller) = StateControllerBuilder::new(
            Arc::new(CounterLogic),
            session.clone(),
            Arc::new(CounterSnapshotter {
                manager: session.manager().clone(),
            }),
            Counter::default(),
        )
        .build();
        tokio::spawn(controller.run());

        let data = web::Data::new(session.clone());
        let server = HttpServer::new(move || {
            App::new()
                .app_data(data.clone())
                .route("/ws", web::get().to(ws_route))
        })
        .workers(1)
        .bind(ADDR)
        .expect("bind");
        ready.send(session).expect("main is waiting");
        server.run().await.expect("serve");
    });
}

struct Client {
    name: &'static str,
    pump: FramePump<JsonCodec>,
    start: Instant,
    values: VecDeque<i64>,
    closed: Option<Closed>,
}

impl Client {
    fn connect(name: &'static str, token: &str) -> Self {
        let pump = FramePump::connect(URL, JsonCodec, PROTOCOL)
            .expect("connect")
            .credential(token);
        Client {
            name,
            pump,
            start: Instant::now(),
            values: VecDeque::new(),
            closed: None,
        }
    }

    fn send(&mut self, op: CounterOp) {
        self.pump.send_op(&op);
    }

    fn poll(&mut self) {
        let mut arrivals = Vec::new();
        self.pump
            .poll(self.start.elapsed().as_millis() as u64, &mut arrivals);
        for arrival in arrivals {
            match arrival {
                Arrival::Ops(frame) => {
                    let ops: Vec<CounterOp> = JsonCodec.decode(frame.body()).expect("decode");
                    for op in ops {
                        if let CounterOp::Value(value) = op {
                            self.values.push_back(value);
                        }
                    }
                }
                Arrival::Closed(closed) => self.closed = Some(closed),
                Arrival::Mismatch { .. } => {
                    println!("{}: the server speaks another wire format", self.name)
                }
                Arrival::Opened => {}
            }
        }
    }

    fn next_value(&mut self) -> i64 {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(value) = self.values.pop_front() {
                println!("{}: value {value}", self.name);
                return value;
            }
            if let Some(closed) = &self.closed {
                panic!("{}: closed: {closed}", self.name);
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: no value within 5s", self.name);
    }

    fn wait_closed(&mut self) {
        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            self.poll();
            if let Some(closed) = &self.closed {
                let detail = String::from_utf8_lossy(closed.detail.as_deref().unwrap_or_default());
                println!(
                    "{}: closed with {:?} ({detail}), refused: {}",
                    self.name,
                    closed.code,
                    closed.refused()
                );
                return;
            }
            thread::sleep(Duration::from_millis(5));
        }
        panic!("{}: still open after 5s", self.name);
    }
}

fn main() {
    let (ready, started) = mpsc::channel();
    thread::spawn(move || serve(ready));
    let session = started.recv().expect("the server started");

    let mut alice = Client::connect("alice", "alice-token");
    alice.next_value();
    alice.send(CounterOp::Increment(5));
    alice.next_value();

    let mut bob = Client::connect("bob", "bob-token");
    bob.next_value();
    bob.send(CounterOp::Increment(2));
    bob.next_value();
    alice.next_value();

    let mut mallory = Client::connect("mallory", "guessed-token");
    mallory.wait_closed();

    let stats = session.stats();
    println!(
        "server: admitted {}, refused {}",
        stats.admitted(),
        stats.refused()
    );

    let mut carol = Client::connect("carol", "carol-token");
    carol.next_value();
    for _ in 0..20 {
        carol.send(CounterOp::Increment(1));
    }
    for _ in 0..5 {
        carol.next_value();
    }
    thread::sleep(Duration::from_millis(500));
    carol.poll();
    println!("carol: {} more values", carol.values.len());

    let shed = session.manager().agent_inbound(&3).shed;
    println!("server: shed {shed} of carol's frames");

    let mut dave = Client::connect("dave", "dave-token");
    dave.next_value();
    session
        .manager()
        .set_agent_outbound_budget(&4, Some(OutboundBudget::frames_per_second(1.5)));
    for _ in 0..5 {
        dave.send(CounterOp::Increment(1));
    }
    dave.next_value();
    dave.next_value();
    thread::sleep(Duration::from_millis(500));
    dave.poll();
    println!("dave: {} more values", dave.values.len());

    thread::sleep(Duration::from_millis(500));
    dave.send(CounterOp::Increment(1));
    dave.next_value();

    let withheld = session.manager().agent_outbound(&4).withheld;
    println!("server: withheld {withheld} frames from dave");
}
```

`budget_outbound` gives every connection thirty frames a second, which none of the earlier clients come near. Dave's link is slow, so once he is connected `set_agent_outbound_budget` gives him one and a half frames a second. A budget banks one second of credit by default and a new one starts full.

Run it:

```shell-session
$ cargo run
alice: value 0
alice: value 5
bob: value 5
bob: value 7
alice: value 7
mallory: closed with Some(4403) (unknown token), refused: true
server: admitted 2, refused 1
carol: value 7
carol: value 8
carol: value 9
carol: value 10
carol: value 11
carol: value 12
carol: 0 more values
server: shed 15 of carol's frames
dave: value 12
dave: value 13
dave: value 14
dave: 0 more values
dave: value 18
server: withheld 3 frames from dave
```

Dave's five increments each ask for a snapshot. Credit is allowed to run negative: a frame goes whenever any credit is left, then the connection waits until the debt is paid off. His 1.5 frames of credit pay for the snapshot at 13 and leave half a frame, which still pays for the one at 14. The other three are withheld and `agent_outbound(&4).withheld` counts them. Nothing more arrives until the credit refills. A second later his next change brings 18, which carries every increment he missed. Skipping a snapshot of absolute state costs latency and never correctness.

Like the rate, there is no default budget. Both are per connection and both are set through `SessionOptions` for everyone or on the manager for one connection.

## Where to go next

The same session works over TCP. `TcpPlazaSession::bind_with_admitter` takes the same admitter, codec and `SessionOptions`. Its frames ride behind a 4-byte big-endian length that `plaza_wire::framing` reads and writes. Your `StateLogic` cannot tell the transports apart, so you can develop in process, test over TCP and ship over WebSockets. The next tutorial uses the same session to impair the link to one client while the program runs.

Guide chapters [30, Bytes on the wire](https://github.com/excsn/plaza/blob/main/docs/guide/30-bytes-on-the-wire.md) and [32, Serving your game](https://github.com/excsn/plaza/blob/main/docs/guide/32-serving-your-game.md) explain the frame format, the version handshake and how the transports are sized. [Chapter 40, The right to say no](https://github.com/excsn/plaza/blob/main/docs/guide/40-the-right-to-say-no.md) covers admission and close codes in depth.

The [plaza_session usage guide](https://github.com/excsn/plaza/blob/main/session/README.USAGE.md) covers workloads, queue policies, deadlines, kicking a connection and hosting a browser page. The [plaza_ws usage guide](https://github.com/excsn/plaza/blob/main/ws_client/README.USAGE.md) covers the browser backend and testing a client without a network. The [plaza_wire usage guide](https://github.com/excsn/plaza/blob/main/wire/README.USAGE.md) covers codecs, the build-time version and packing bits by hand.

The full API references are on docs.rs: [plaza_session](https://docs.rs/plaza_session), [plaza_ws](https://docs.rs/plaza_ws) and [plaza_wire](https://docs.rs/plaza_wire).