LLD in Rust: traits, enums and ownership

session · door_session.py

from session import run, version

version()
run(
    """
rustc --edition 2024 -D warnings door/door.rs -o door_demo
./door_demo
"""
)

# rustc 1.99.0 (b940084d7)
$ rustc --edition 2024 -D warnings door/door.rs -o door_demo
$ ./door_demo
Closed --Press--> Opening { percent: 0 }
Opening { percent: 0 } --Tick--> Opening { percent: 50 }
Obstacle refused while Opening { percent: 50 }
Opening { percent: 50 } --Tick--> Open { seconds_left: 2 }
    ding at floor 3
Open { seconds_left: 2 } --Tick--> Open { seconds_left: 1 }
Open { seconds_left: 1 } --Tick--> Open { seconds_left: 0 }
Open { seconds_left: 0 } --Tick--> Closing { percent: 100 }
    doors closing
Closing { percent: 100 } --Obstacle--> Opening { percent: 0 }

door/door.rs

// A lift door in Rust: its states are an enum whose variants carry their own data, every move goes
// through one exhaustive match that returns a Result, and the people who must hear about a move are
// trait objects that the lift owns.
use std::fmt;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Door {
    Closed,
    Opening { percent: u8 },
    Open { seconds_left: u8 },
    Closing { percent: u8 },
}

#[derive(Debug, Clone, Copy)]
enum Event {
    Press,
    Tick,
    Obstacle,
}

#[derive(Debug)]
struct Refused {
    door: Door,
    event: Event,
}

impl fmt::Display for Refused {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{:?} refused while {:?}", self.event, self.door)
    }
}

fn step(door: Door, event: Event) -> Result<Door, Refused> {
    use Door::*;
    match (door, event) {
        (Closed, Event::Press) => Ok(Opening { percent: 0 }),
        (Closed, Event::Tick) => Ok(Closed),
        (Opening { percent }, Event::Tick) if percent >= 50 => Ok(Open { seconds_left: 2 }),
        (Opening { percent }, Event::Tick) => Ok(Opening { percent: percent + 50 }),
        (Open { .. }, Event::Press) => Ok(Open { seconds_left: 2 }),
        (Open { seconds_left: 0 }, Event::Tick) => Ok(Closing { percent: 100 }),
        (Open { seconds_left }, Event::Tick) => Ok(Open { seconds_left: seconds_left - 1 }),
        (Closing { .. }, Event::Obstacle | Event::Press) => Ok(Opening { percent: 0 }),
        (Closing { percent }, Event::Tick) if percent <= 50 => Ok(Closed),
        (Closing { percent }, Event::Tick) => Ok(Closing { percent: percent - 50 }),
        (door @ (Closed | Opening { .. } | Open { .. }), event @ (Event::Obstacle | Event::Press)) => {
            Err(Refused { door, event })
        }
    }
}

// Anything that must hear about the door implements this trait; the lift does not know which.
trait Announcer {
    fn announce(&self, floor: FloorId, door: Door) -> Option<String>;
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct FloorId(u8);

struct Chime;
impl Announcer for Chime {
    fn announce(&self, floor: FloorId, door: Door) -> Option<String> {
        matches!(door, Door::Open { seconds_left: 2 }).then(|| format!("ding at floor {}", floor.0))
    }
}

struct Panel;
impl Announcer for Panel {
    fn announce(&self, _floor: FloorId, door: Door) -> Option<String> {
        matches!(door, Door::Closing { percent: 100 }).then(|| "doors closing".to_string())
    }
}

struct Lift {
    floor: FloorId,
    door: Door,
    announcers: Vec<Box<dyn Announcer>>,
}

impl Lift {
    fn handle(&mut self, event: Event) {
        match step(self.door, event) {
            Ok(next) => {
                println!("{:?} --{:?}--> {:?}", self.door, event, next);
                self.door = next;
                for a in &self.announcers {
                    if let Some(text) = a.announce(self.floor, next) {
                        println!("    {text}");
                    }
                }
            }
            Err(refused) => println!("{refused}"),
        }
    }
}

fn main() {
    let mut lift = Lift { floor: FloorId(3), door: Door::Closed, announcers: vec![Box::new(Chime), Box::new(Panel)] };
    use Event::*;
    for event in [Press, Tick, Obstacle, Tick, Tick, Tick, Tick, Obstacle] {
        lift.handle(event);
    }
}

session.py

"""Prints a terminal session: "$ " and each command, then everything it printed (standard output and standard
error together), and its exit status when that is not 0.

`rustc` is Rust 1.99.0, the version this lesson pins, from its rustup toolchain in the user's home folder.
learn:verify's sandbox may not read that folder, so these sessions are recorded with LEARN_VERIFY_NETWORK=allow. They
read only this folder and the toolchain, write only inside this folder, and never use the network.

A command that fails makes the session exit with status 1, unless the script lists it in may_fail. For a command
that fails, `keep` (a regular expression) shows only the lines of its output that matter: the compiler's error and
the "-->" line with its location. Version lines are shown without their build date, and test timings, which change on
every run, are left out.
"""

import os
import pwd
import re
import subprocess
import sys
from pathlib import Path

RUST = Path(pwd.getpwuid(os.getuid()).pw_dir) / ".rustup" / "toolchains" / "1.99.0-aarch64-apple-darwin" / "bin"
BUILD_DATE = re.compile(r"\s+\d{4}-\d{2}-\d{2}(?=\))")
TIMING = re.compile(r"; finished in \d+\.\d+s")


def _env():
    here = Path.cwd()
    for folder in ("home", "tmp"):
        (here / folder).mkdir(exist_ok=True)
    env = dict(os.environ)
    env.update(
        PATH=os.pathsep.join([str(RUST), "/usr/bin", "/bin"]),
        HOME=str(here / "home"),
        TMPDIR=str(here / "tmp"),
        RUST_BACKTRACE="0",
        NO_COLOR="1",
    )
    return env


def _shell(command):
    done = subprocess.run(command, shell=True, env=_env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    return done.returncode, [TIMING.sub("", line) for line in done.stdout.splitlines()]


def version():
    """Prints `rustc --version` as a comment, without its build date."""
    status, lines = _shell("rustc --version")
    if status != 0 or not lines:
        sys.exit("Rust 1.99.0 is not installed on this computer")
    print("#", BUILD_DATE.sub("", lines[0]))


def _kept(lines, keep):
    out = []
    for i, line in enumerate(lines):
        if re.search(keep, line):
            out.append(line)
            if i + 1 < len(lines) and lines[i + 1].lstrip().startswith("--> "):
                out.append(lines[i + 1])
    return out


def run(script, may_fail=(), keep=None):
    """Runs each line of the script as one shell command, in the examples folder."""
    unexpected = 0
    for command in (line.strip() for line in script.strip().splitlines()):
        print(f"$ {command}")
        status, lines = _shell(command)
        if status != 0 and keep:
            lines = _kept(lines, keep)
        for line in lines:
            print(line)
        if status != 0:
            print(f"[exit status {status}]")
            if command not in may_fail:
                unexpected += 1
    if unexpected:
        sys.exit(1)