Run a turn-based game

For Rust developers who have built a plaza controller and want turns, rounds and phases without writing the bookkeeping by hand

You will build a two-player game called high number. The players take turns to play a number and the highest number wins the round. The game waits in a lobby until someone starts it, refuses moves made out of turn and ends after two rounds. It uses the game_common::flow_control module from the plaza crate: RoundRobinTurnManager for the turn order, Phased for the stage the game is in and SequentialRoundManager for the rounds. An OpGuard decides which ops are legal at each point.

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 high_number
cd high_number

Your [dependencies] section should look like this:

Cargo.toml

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

Flow control is part of the plaza crate and needs no feature flag. The programs use Tokio's single-threaded current_thread runtime and send each player's op straight to the controller with ControllerCommand::SubmitAgentOps. That keeps the moves in a fixed order, so the output is the same on every run. A real client sends its ops through the session and the guard and the rules treat them the same way.

Seat players in a turn order#

RoundRobinTurnManager holds the players in order and knows whose turn it is. begin gives the first player the turn and end_current_turn_and_advance passes it to the next, wrapping round at the end. Every change queues a TurnChangedNoticePayload for all clients, so a client always knows whose turn it is without anyone writing that broadcast.

The manager cannot know your op type, so you hand it a function that wraps the notice into one. GameOp::TurnChanged is that function: a tuple variant's name works as a function from its field to the enum. The notices go into a context you pass in. OpsQueue is the minimal one. ctx.into_ops() turns what it collected into the rules' output. Ops of your own go into the same queue through ctx.ops_q(), which needs the FsmContext trait in scope, so everything leaves in the order it happened.

Replace src/main.rs with this:

src/main.rs

use std::sync::Arc;

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    common::fsm::{FsmContext, OpsQueue},
    controller::{CommandSender, ControllerCommand, StateControllerBuilder, query_with},
    game_common::flow_control::{
        RoundRobinTurnManager, TurnManager, turns::op_payloads::TurnChangedNoticePayload,
    },
    session::{InProcessSession, TargetedOp, in_process::ClientInbox},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
};

type PlayerId = u32;

const ALICE: PlayerId = 1;
const BOB: PlayerId = 2;
const TABLE: PlayerId = 9;

#[derive(Debug)]
struct Game {
    turns: RoundRobinTurnManager<GameOp, PlayerId, PlayerId>,
}

#[derive(Clone, Debug)]
enum GameOp {
    Start,
    Play(u32),
    Played { player: PlayerId, value: u32 },
    TurnChanged(TurnChangedNoticePayload<PlayerId>),
}

fn to_all(op: GameOp) -> TargetedOp<GameOp, PlayerId> {
    TargetedOp::new_system_all(vec![op])
}

struct GameRules;

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

        let mut ctx = OpsQueue::new();
        for op in ops {
            match op {
                GameOp::Start => {
                    state.turns.begin(&mut ctx);
                }
                GameOp::Play(value) => {
                    ctx.ops_q().push(to_all(GameOp::Played { player, value }));
                    state
                        .turns
                        .end_current_turn_and_advance(&mut ctx)
                        .map_err(StateLogicError::PreconditionFailed)?;
                }
                _ => {}
            }
        }
        Ok(ctx.into_ops().into())
    }
}

type Tx = CommandSender<GameOp, PlayerId, Game>;

async fn send(tx: &Tx, player: PlayerId, op: GameOp) -> Result<(), Box<dyn std::error::Error>> {
    tx.send(ControllerCommand::SubmitAgentOps {
        agent: Agent::new_human(player),
        ops: vec![op],
    })
    .await?;
    Ok(())
}

async fn look(
    tx: &Tx,
    table: &mut ClientInbox<GameOp, PlayerId>,
) -> Result<(), Box<dyn std::error::Error>> {
    query_with(tx, |_| ()).await?;
    while let Ok(msg) = table.try_recv() {
        for op in msg.ops {
            match op {
                GameOp::Played { player, value } => {
                    println!("table: player {player} played {value}")
                }
                GameOp::TurnChanged(turn) => {
                    if let Some(player) = turn.new_turn_actor {
                        println!("table: turn {} is player {player}'s", turn.turn_number);
                    }
                }
                _ => {}
            }
        }
    }
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let game = Game {
        turns: RoundRobinTurnManager::new(vec![ALICE, BOB], GameOp::TurnChanged),
    };
    let session = InProcessSession::<GameOp, PlayerId>::new();
    let (tx, controller) =
        StateControllerBuilder::without_snapshots(Arc::new(GameRules), session.clone(), game)
            .build();
    tokio::spawn(controller.run());

    let (_, mut table) = session.connect(Agent::new_human(TABLE)).await?;

    send(&tx, ALICE, GameOp::Start).await?;
    look(&tx, &mut table).await?;
    for (player, value) in [(ALICE, 5), (BOB, 7), (BOB, 3)] {
        send(&tx, player, GameOp::Play(value)).await?;
        look(&tx, &mut table).await?;
    }

    Ok(())
}

The game state holds the turn manager and nothing else yet. A spectator connects as TABLE and look prints what it sees. look first sends the controller an empty query_with. The query travels the same command queue as the moves, so by the time it answers, every move sent before it has been handled and its broadcasts are already in the inbox for try_recv to drain.

Run it:

$ cargo run
table: turn 1 is player 1's
table: player 1 played 5
table: turn 2 is player 2's
table: player 2 played 7
table: turn 3 is player 1's
table: player 2 played 3
table: turn 4 is player 2's

The turns alternate as they should, but the last move was Bob's on Alice's turn and the rules accepted it. Nothing in the rules checks whose turn it is, so the turn manager passed the turn along as if Alice had played.

Make the phase decide what is legal#

Whether a player may make a move at all is a separate question from what the move does. plaza puts the first question in an OpGuard. The controller runs the guard on every op from an agent before the rules see it, with the state read-only. A refused op never reaches the rules. The guard can also send the player a reply saying why.

The game also needs a stage. Before it starts, nobody may play. Once it has started, only the player whose turn it is may play and nobody may start it again. Phased<Stage> holds the stage. The only way to change it is transition_to, which also queues a notice for every client, the same way the turn manager does.

Add phases::op_payloads::PhaseChangedNoticePayload and Phased to the flow_control import and op_guard::{GuardFn, OpClearance} to the plaza list. The stage goes into the state and two variants join the op enum: Refused for the guard's replies and StageChanged for the stage notices.

src/main.rs

#[derive(Clone, Debug, PartialEq)]
enum Stage {
    Lobby,
    Playing,
}

#[derive(Debug)]
struct Game {
    stage: Phased<Stage>,
    turns: RoundRobinTurnManager<GameOp, PlayerId, PlayerId>,
}

#[derive(Clone, Debug)]
enum GameOp {
    Start,
    Play(u32),
    Refused(String),
    Played { player: PlayerId, value: u32 },
    TurnChanged(TurnChangedNoticePayload<PlayerId>),
    StageChanged(PhaseChangedNoticePayload<Stage>),
}

The guard is a plain function. It matches on the stage and the op together. The turn check is one more condition on the Playing arm:

src/main.rs

fn referee(game: &Game, source: &Agent<PlayerId>, op: &GameOp) -> OpClearance<GameOp> {
    let refuse = |why: &str| OpClearance::Refused {
        reply: Some(GameOp::Refused(why.to_string())),
    };
    match (game.stage.current(), op) {
        (Stage::Lobby, GameOp::Start) => OpClearance::Cleared,
        (Stage::Lobby, GameOp::Play(_)) => refuse("the game has not started"),
        (Stage::Playing, GameOp::Play(_))
            if game.turns.current_turn_actor().as_ref() == source.id() =>
        {
            OpClearance::Cleared
        }
        (Stage::Playing, GameOp::Play(_)) => refuse("it is not your turn"),
        _ => refuse("you cannot do that now"),
    }
}

The last arm refuses everything else, which includes a client trying to send an op only the server should send, such as Played.

The rules no longer check anything. Start moves the stage to Playing and begins the turns:

src/main.rs

fn to_all(op: GameOp) -> TargetedOp<GameOp, PlayerId> {
    TargetedOp::new_system_all(vec![op])
}

struct GameRules;

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

        let mut ctx = OpsQueue::new();
        for op in ops {
            match op {
                GameOp::Start => {
                    state
                        .stage
                        .transition_to(Stage::Playing, &mut ctx, GameOp::StageChanged);
                    state.turns.begin(&mut ctx);
                }
                GameOp::Play(value) => {
                    ctx.ops_q().push(to_all(GameOp::Played { player, value }));
                    state
                        .turns
                        .end_current_turn_and_advance(&mut ctx)
                        .map_err(StateLogicError::PreconditionFailed)?;
                }
                _ => {}
            }
        }
        Ok(ctx.into_ops().into())
    }
}

look now also prints refusals. Those go only to the player who made the move, so it reads each player's inbox for them:

src/main.rs

async fn look(
    tx: &Tx,
    table: &mut ClientInbox<GameOp, PlayerId>,
    players: &mut [(&str, &mut ClientInbox<GameOp, PlayerId>)],
) -> Result<(), Box<dyn std::error::Error>> {
    query_with(tx, |_| ()).await?;
    for (name, inbox) in players.iter_mut() {
        while let Ok(msg) = inbox.try_recv() {
            for op in msg.ops {
                if let GameOp::Refused(why) = op {
                    println!("{name}: refused, {why}");
                }
            }
        }
    }
    while let Ok(msg) = table.try_recv() {
        for op in msg.ops {
            match op {
                GameOp::Played { player, value } => {
                    println!("table: player {player} played {value}")
                }
                GameOp::TurnChanged(turn) => {
                    if let Some(player) = turn.new_turn_actor {
                        println!("table: turn {} is player {player}'s", turn.turn_number);
                    }
                }
                GameOp::StageChanged(notice) => {
                    println!("table: stage is now {:?}", notice.new_phase)
                }
                _ => {}
            }
        }
    }
    Ok(())
}

In main, the builder takes the guard wrapped in GuardFn. Every player connects so they can receive their refusals:

src/main.rs

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let game = Game {
        stage: Phased::new(Stage::Lobby),
        turns: RoundRobinTurnManager::new(vec![ALICE, BOB], GameOp::TurnChanged),
    };
    let session = InProcessSession::<GameOp, PlayerId>::new();
    let (tx, controller) =
        StateControllerBuilder::without_snapshots(Arc::new(GameRules), session.clone(), game)
            .guard(Arc::new(GuardFn(referee)))
            .build();
    tokio::spawn(controller.run());

    let (_, mut table) = session.connect(Agent::new_human(TABLE)).await?;
    let (_, mut alice) = session.connect(Agent::new_human(ALICE)).await?;
    let (_, mut bob) = session.connect(Agent::new_human(BOB)).await?;

    let moves = [
        (BOB, GameOp::Play(4)),
        (ALICE, GameOp::Start),
        (ALICE, GameOp::Play(5)),
        (BOB, GameOp::Play(7)),
        (BOB, GameOp::Play(3)),
        (BOB, GameOp::Start),
    ];
    for (player, op) in moves {
        send(&tx, player, op).await?;
        look(
            &tx,
            &mut table,
            &mut [("alice", &mut alice), ("bob", &mut bob)],
        )
        .await?;
    }

    Ok(())
}

Run it:

$ cargo run
bob: refused, the game has not started
table: stage is now Playing
table: turn 1 is player 1's
table: player 1 played 5
table: turn 2 is player 2's
table: player 2 played 7
table: turn 3 is player 1's
bob: refused, it is not your turn
bob: refused, you cannot do that now

The guard refused three of Bob's moves: playing in the lobby, playing out of turn and starting a game already running. None of them reached the rules, so the turn order is still right. Without a reply, a refused op would look to the player like the game had frozen. ControllerStats::ops_refused counts refusals if you want to watch for clients that keep trying.

The guard is synchronous because it runs on the controller's task for every op. Anything it needs to decide has to be in the state already.

End a round#

A round ends once every player has played. SequentialRoundManager counts the rounds and announces each start and end. start_next_round begins the next one and returns an error if a round is still running or the last one has been played. end_round_with ends it with a reason and an optional summary of your own type, which reaches clients in the notice. is_finished says whether the round limit has been reached.

The state gains the round manager, the moves played this round and a Finished stage. The op enum gains a variant for each round notice:

src/main.rs

#[derive(Clone, Debug, PartialEq)]
enum Stage {
    Lobby,
    Playing,
    Finished,
}

#[derive(Clone, Debug)]
struct RoundSummary {
    winner: PlayerId,
    value: u32,
}

#[derive(Debug)]
struct Game {
    stage: Phased<Stage>,
    turns: RoundRobinTurnManager<GameOp, PlayerId, PlayerId>,
    rounds: SequentialRoundManager<GameOp, PlayerId, RoundSummary>,
    played: Vec<(PlayerId, u32)>,
}

#[derive(Clone, Debug)]
enum GameOp {
    Start,
    Play(u32),
    Refused(String),
    Played { player: PlayerId, value: u32 },
    TurnChanged(TurnChangedNoticePayload<PlayerId>),
    StageChanged(PhaseChangedNoticePayload<Stage>),
    RoundStarted(RoundStartedNoticePayload),
    RoundEnded(RoundEndedNoticePayload<RoundSummary>),
}

The guard gets one more arm: once the game is over, everything is refused.

src/main.rs

fn referee(game: &Game, source: &Agent<PlayerId>, op: &GameOp) -> OpClearance<GameOp> {
    let refuse = |why: &str| OpClearance::Refused {
        reply: Some(GameOp::Refused(why.to_string())),
    };
    match (game.stage.current(), op) {
        (Stage::Lobby, GameOp::Start) => OpClearance::Cleared,
        (Stage::Lobby, GameOp::Play(_)) => refuse("the game has not started"),
        (Stage::Playing, GameOp::Play(_))
            if game.turns.current_turn_actor().as_ref() == source.id() =>
        {
            OpClearance::Cleared
        }
        (Stage::Playing, GameOp::Play(_)) => refuse("it is not your turn"),
        (Stage::Finished, _) => refuse("the game is over"),
        _ => refuse("you cannot do that now"),
    }
}

The rules now count the moves. When the last player in the order has played, the round ends instead of the turn passing on. Then either the next round starts or the game is over. restart gives the turn back to the first player for the new round. begin would not do that, since it leaves a turn already in progress alone.

src/main.rs

fn to_all(op: GameOp) -> TargetedOp<GameOp, PlayerId> {
    TargetedOp::new_system_all(vec![op])
}

fn start_round(
    state: &mut Game,
    ctx: &mut OpsQueue<GameOp, PlayerId>,
) -> Result<(), StateLogicError> {
    state.played.clear();
    state
        .rounds
        .start_next_round(ctx)
        .map_err(StateLogicError::PreconditionFailed)?;
    state.turns.restart(ctx);
    Ok(())
}

fn end_round(
    state: &mut Game,
    ctx: &mut OpsQueue<GameOp, PlayerId>,
) -> Result<(), StateLogicError> {
    let Some(&(winner, value)) = state.played.iter().max_by_key(|(_, value)| *value) else {
        return Ok(());
    };
    let summary = RoundSummary { winner, value };
    state
        .rounds
        .end_round_with(ctx, "everyone has played", Some(summary));
    if state.rounds.is_finished() {
        state
            .stage
            .transition_to(Stage::Finished, ctx, GameOp::StageChanged);
        Ok(())
    } else {
        start_round(state, ctx)
    }
}

struct GameRules;

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

        let mut ctx = OpsQueue::new();
        for op in ops {
            match op {
                GameOp::Start => {
                    state
                        .stage
                        .transition_to(Stage::Playing, &mut ctx, GameOp::StageChanged);
                    start_round(state, &mut ctx)?;
                }
                GameOp::Play(value) => {
                    state.played.push((player, value));
                    ctx.ops_q().push(to_all(GameOp::Played { player, value }));
                    if state.played.len() == state.turns.actors().len() {
                        end_round(state, &mut ctx)?;
                    } else {
                        state
                            .turns
                            .end_current_turn_and_advance(&mut ctx)
                            .map_err(StateLogicError::PreconditionFailed)?;
                    }
                }
                _ => {}
            }
        }
        Ok(ctx.into_ops().into())
    }
}

main builds the round manager with a limit of two rounds and plays both of them:

src/main.rs

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let game = Game {
        stage: Phased::new(Stage::Lobby),
        turns: RoundRobinTurnManager::new(vec![ALICE, BOB], GameOp::TurnChanged),
        rounds: SequentialRoundManager::new(Some(ROUNDS), GameOp::RoundStarted, GameOp::RoundEnded),
        played: Vec::new(),
    };
    let session = InProcessSession::<GameOp, PlayerId>::new();
    let (tx, controller) =
        StateControllerBuilder::without_snapshots(Arc::new(GameRules), session.clone(), game)
            .guard(Arc::new(GuardFn(referee)))
            .build();
    tokio::spawn(controller.run());

    let (_, mut table) = session.connect(Agent::new_human(TABLE)).await?;
    let (_, mut alice) = session.connect(Agent::new_human(ALICE)).await?;
    let (_, mut bob) = session.connect(Agent::new_human(BOB)).await?;

    let moves = [
        (ALICE, GameOp::Start),
        (ALICE, GameOp::Play(5)),
        (BOB, GameOp::Play(7)),
        (ALICE, GameOp::Play(9)),
        (BOB, GameOp::Play(2)),
        (BOB, GameOp::Play(8)),
    ];
    for (player, op) in moves {
        send(&tx, player, op).await?;
        look(
            &tx,
            &mut table,
            &mut [("alice", &mut alice), ("bob", &mut bob)],
        )
        .await?;
    }

    Ok(())
}

look gains arms for the round notices, which are in the full listing below.

Run it:

$ cargo run
table: stage is now Playing
table: round 1 begins
table: turn 1 is player 1's
table: player 1 played 5
table: turn 2 is player 2's
table: player 2 played 7
table: round 1 goes to player 2 with 7
table: round 2 begins
table: turn 1 is player 1's
table: player 1 played 9
table: turn 2 is player 2's
table: player 2 played 2
table: round 2 goes to player 1 with 9
table: stage is now Finished
bob: refused, the game is over

Each round starts with Alice on turn 1, because restart seats the first player again and resets the turn number. After the second round is_finished was true, so the stage moved to Finished and the guard refused Bob's next move.

Here is the whole program:

src/main.rs

use std::sync::Arc;

use async_trait::async_trait;
use plaza::{
    agent::Agent,
    common::fsm::{FsmContext, OpsQueue},
    controller::{CommandSender, ControllerCommand, StateControllerBuilder, query_with},
    game_common::flow_control::{
        Phased, RoundManager, RoundRobinTurnManager, SequentialRoundManager, TurnManager,
        phases::op_payloads::PhaseChangedNoticePayload,
        rounds::op_payloads::{RoundEndedNoticePayload, RoundStartedNoticePayload},
        turns::op_payloads::TurnChangedNoticePayload,
    },
    op_guard::{GuardFn, OpClearance},
    session::{InProcessSession, TargetedOp, in_process::ClientInbox},
    state_logic::{LogicInput, LogicOutput, StateLogic, StateLogicError},
};

type PlayerId = u32;

const ALICE: PlayerId = 1;
const BOB: PlayerId = 2;
const TABLE: PlayerId = 9;
const ROUNDS: u32 = 2;

#[derive(Clone, Debug, PartialEq)]
enum Stage {
    Lobby,
    Playing,
    Finished,
}

#[derive(Clone, Debug)]
struct RoundSummary {
    winner: PlayerId,
    value: u32,
}

#[derive(Debug)]
struct Game {
    stage: Phased<Stage>,
    turns: RoundRobinTurnManager<GameOp, PlayerId, PlayerId>,
    rounds: SequentialRoundManager<GameOp, PlayerId, RoundSummary>,
    played: Vec<(PlayerId, u32)>,
}

#[derive(Clone, Debug)]
enum GameOp {
    Start,
    Play(u32),
    Refused(String),
    Played { player: PlayerId, value: u32 },
    TurnChanged(TurnChangedNoticePayload<PlayerId>),
    StageChanged(PhaseChangedNoticePayload<Stage>),
    RoundStarted(RoundStartedNoticePayload),
    RoundEnded(RoundEndedNoticePayload<RoundSummary>),
}

fn referee(game: &Game, source: &Agent<PlayerId>, op: &GameOp) -> OpClearance<GameOp> {
    let refuse = |why: &str| OpClearance::Refused {
        reply: Some(GameOp::Refused(why.to_string())),
    };
    match (game.stage.current(), op) {
        (Stage::Lobby, GameOp::Start) => OpClearance::Cleared,
        (Stage::Lobby, GameOp::Play(_)) => refuse("the game has not started"),
        (Stage::Playing, GameOp::Play(_))
            if game.turns.current_turn_actor().as_ref() == source.id() =>
        {
            OpClearance::Cleared
        }
        (Stage::Playing, GameOp::Play(_)) => refuse("it is not your turn"),
        (Stage::Finished, _) => refuse("the game is over"),
        _ => refuse("you cannot do that now"),
    }
}

fn to_all(op: GameOp) -> TargetedOp<GameOp, PlayerId> {
    TargetedOp::new_system_all(vec![op])
}

fn start_round(
    state: &mut Game,
    ctx: &mut OpsQueue<GameOp, PlayerId>,
) -> Result<(), StateLogicError> {
    state.played.clear();
    state
        .rounds
        .start_next_round(ctx)
        .map_err(StateLogicError::PreconditionFailed)?;
    state.turns.restart(ctx);
    Ok(())
}

fn end_round(
    state: &mut Game,
    ctx: &mut OpsQueue<GameOp, PlayerId>,
) -> Result<(), StateLogicError> {
    let Some(&(winner, value)) = state.played.iter().max_by_key(|(_, value)| *value) else {
        return Ok(());
    };
    let summary = RoundSummary { winner, value };
    state
        .rounds
        .end_round_with(ctx, "everyone has played", Some(summary));
    if state.rounds.is_finished() {
        state
            .stage
            .transition_to(Stage::Finished, ctx, GameOp::StageChanged);
        Ok(())
    } else {
        start_round(state, ctx)
    }
}

struct GameRules;

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

        let mut ctx = OpsQueue::new();
        for op in ops {
            match op {
                GameOp::Start => {
                    state
                        .stage
                        .transition_to(Stage::Playing, &mut ctx, GameOp::StageChanged);
                    start_round(state, &mut ctx)?;
                }
                GameOp::Play(value) => {
                    state.played.push((player, value));
                    ctx.ops_q().push(to_all(GameOp::Played { player, value }));
                    if state.played.len() == state.turns.actors().len() {
                        end_round(state, &mut ctx)?;
                    } else {
                        state
                            .turns
                            .end_current_turn_and_advance(&mut ctx)
                            .map_err(StateLogicError::PreconditionFailed)?;
                    }
                }
                _ => {}
            }
        }
        Ok(ctx.into_ops().into())
    }
}

type Tx = CommandSender<GameOp, PlayerId, Game>;

async fn send(tx: &Tx, player: PlayerId, op: GameOp) -> Result<(), Box<dyn std::error::Error>> {
    tx.send(ControllerCommand::SubmitAgentOps {
        agent: Agent::new_human(player),
        ops: vec![op],
    })
    .await?;
    Ok(())
}

async fn look(
    tx: &Tx,
    table: &mut ClientInbox<GameOp, PlayerId>,
    players: &mut [(&str, &mut ClientInbox<GameOp, PlayerId>)],
) -> Result<(), Box<dyn std::error::Error>> {
    query_with(tx, |_| ()).await?;
    for (name, inbox) in players.iter_mut() {
        while let Ok(msg) = inbox.try_recv() {
            for op in msg.ops {
                if let GameOp::Refused(why) = op {
                    println!("{name}: refused, {why}");
                }
            }
        }
    }
    while let Ok(msg) = table.try_recv() {
        for op in msg.ops {
            match op {
                GameOp::Played { player, value } => {
                    println!("table: player {player} played {value}")
                }
                GameOp::TurnChanged(turn) => {
                    if let Some(player) = turn.new_turn_actor {
                        println!("table: turn {} is player {player}'s", turn.turn_number);
                    }
                }
                GameOp::StageChanged(notice) => {
                    println!("table: stage is now {:?}", notice.new_phase)
                }
                GameOp::RoundStarted(round) => {
                    println!("table: round {} begins", round.round_number)
                }
                GameOp::RoundEnded(round) => {
                    if let Some(summary) = round.summary_data {
                        println!(
                            "table: round {} goes to player {} with {}",
                            round.round_number, summary.winner, summary.value
                        );
                    }
                }
                _ => {}
            }
        }
    }
    Ok(())
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let game = Game {
        stage: Phased::new(Stage::Lobby),
        turns: RoundRobinTurnManager::new(vec![ALICE, BOB], GameOp::TurnChanged),
        rounds: SequentialRoundManager::new(Some(ROUNDS), GameOp::RoundStarted, GameOp::RoundEnded),
        played: Vec::new(),
    };
    let session = InProcessSession::<GameOp, PlayerId>::new();
    let (tx, controller) =
        StateControllerBuilder::without_snapshots(Arc::new(GameRules), session.clone(), game)
            .guard(Arc::new(GuardFn(referee)))
            .build();
    tokio::spawn(controller.run());

    let (_, mut table) = session.connect(Agent::new_human(TABLE)).await?;
    let (_, mut alice) = session.connect(Agent::new_human(ALICE)).await?;
    let (_, mut bob) = session.connect(Agent::new_human(BOB)).await?;

    let moves = [
        (ALICE, GameOp::Start),
        (ALICE, GameOp::Play(5)),
        (BOB, GameOp::Play(7)),
        (ALICE, GameOp::Play(9)),
        (BOB, GameOp::Play(2)),
        (BOB, GameOp::Play(8)),
    ];
    for (player, op) in moves {
        send(&tx, player, op).await?;
        look(
            &tx,
            &mut table,
            &mut [("alice", &mut alice), ("bob", &mut bob)],
        )
        .await?;
    }

    Ok(())
}

Where to go next#

The turn manager also handles players leaving: remove_actor takes a player out of the order. If it was their turn, play passes to whoever fills their slot instead of stalling. end_current_turn_and_advance returns an Advanced, which tells you whether that move closed a pass over the whole order. A turn clock that plays for a stalled player is scheduled work, which the scheduler tutorial covers with PhasedScheduler and Mark. RoundRobinTurnManager::with_time_limit puts the limit in the turn notice so clients can draw a countdown.

The card_table example plays tricks with turns, rounds, phases and hidden hands. night_watch runs an open-ended game of day and night phases and authorizes ops with a stateful OpGuard. check_raise is a poker betting round, where a raise reopens the order and turn clocks are dropped with a Situation. The plaza guide's The right to say no covers guards alongside admission and kicks.

The plaza usage guide on GitHub covers guards and the optional modules. The full API reference is on docs.rs.