nstreams-core 0.3.1

Generic versioned event stream handler with history replay
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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

use nstreams_core::namespace::StreamNamespace;
use nstreams_core::store::EventStore;
use nstreams_core::stream::ReadStreamBackend;
use nstreams_core::testing::{payload, TestHarness};

#[tokio::test]
async fn worker_persists_without_piping_before_stream_ready() {
    let harness = TestHarness::new();
    let namespace = StreamNamespace::from("events.worker");
    let worker = harness.worker();
    let consumer = harness.write_queue.clone().consumer();

    harness
        .publisher
        .publish(&namespace, &payload(serde_json::json!({"queued": true})), None)
        .await
        .unwrap();

    tokio::spawn(async move {
        worker.run(consumer).await.unwrap();
    });

    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    let version = harness.store.latest_version(&namespace).await.unwrap();
    assert_eq!(version, Some(1));
    assert!(!harness.read_stream.stream_exists(&namespace).await.unwrap());
}

#[tokio::test]
async fn worker_pipes_to_read_stream_after_subscribe() {
    let harness = TestHarness::new();
    let namespace = StreamNamespace::from("events.pipe");
    let worker = harness.worker();
    let consumer = harness.write_queue.clone().consumer();

    tokio::spawn(async move {
        worker.run(consumer).await.unwrap();
    });

    harness.handler.subscribe(&namespace, 0).await.unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    harness
        .publisher
        .publish(&namespace, &payload(serde_json::json!({"piped": true})), None)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    let version = harness.store.latest_version(&namespace).await.unwrap();
    assert_eq!(version, Some(1));
    assert!(harness.handler.is_read_stream_ready(&namespace).await);
}

#[tokio::test]
async fn publish_assigns_monotonic_versions() {
    let harness = TestHarness::new();
    let namespace = StreamNamespace::from("events.versions");

    let first = harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!(1)), None)
        .await
        .unwrap();
    let second = harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!(2)), None)
        .await
        .unwrap();

    assert_eq!(first.version, 1);
    assert_eq!(second.version, 2);
}