edfsm-machine 0.12.0

Tasks and Channels for Event Driven Finite State Machines
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

pub mod fixtures;
use edfsm::Input;
use edfsm_machine::{error::Result, machine, Machine};
use fixtures::{Command, Counter, Event};
use streambed_codec::{Cbor, CommitLogExt};
use streambed_logged::FileLog;
use tokio::{spawn, sync::mpsc::Sender, task::JoinSet};

const TEST_DATA: &str = "test_data";
const TOPIC: &str = "event_series";

async fn producer(sender: Sender<Input<Command, Event>>) -> Result<()> {
    for _ in 1..100 {
        sender.send(Input::Event(Event::Tick)).await?;
    }
    sender.send(Input::Command(Command::Assert(99))).await?;
    Ok(())
}

async fn phase_1() {
    let topic_file = [TEST_DATA, TOPIC].join("/");
    let _ = std::fs::remove_file(&topic_file);
    let _ = std::fs::create_dir(TEST_DATA);

    let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, Cbor);
    let machine = machine::<Counter>().with_event_log(log);
    let prod_task = producer(machine.input());

    let mut set = JoinSet::new();
    set.spawn(machine.task());
    set.spawn(prod_task);
    set.join_all().await;
}

async fn phase_2() {
    let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, Cbor);
    let machine = machine::<Counter>().with_event_log(log);
    let sender = machine.input();
    let handle = spawn(machine.task());
    sender
        .send(Input::Command(Command::Assert(99)))
        .await
        .unwrap();
    drop(sender);
    handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn streambed_test() {
    phase_1().await;
    phase_2().await;
}