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:

plaza = "0"

Set up the project#

Start a new binary crate.

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:

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:

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:

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:

$ 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:

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:

$ 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:

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:

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:

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:

$ 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:

src/main.rs

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:

$ 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:

src/main.rs

use plaza_session::{
    ActixWsPlazaSession, ConnectionAdmission, ConnectionAdmitter, Farewell, Peer, Rate,
    SessionOptions,
};

In serve, build the session from options:

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:

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:

$ 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:

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:

src/main.rs

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:

$ 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 and 32, Serving your game explain the frame format, the version handshake and how the transports are sized. Chapter 40, The right to say no covers admission and close codes in depth.

The plaza_session usage guide covers workloads, queue policies, deadlines, kicking a connection and hosting a browser page. The plaza_ws usage guide covers the browser backend and testing a client without a network. The plaza_wire usage guide covers codecs, the build-time version and packing bits by hand.

The full API references are on docs.rs: plaza_session, plaza_ws and plaza_wire.