nstreams-core 0.1.0

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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

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

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

    harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!({"n": 1})))
        .await
        .unwrap();
    harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!({"n": 2})))
        .await
        .unwrap();

    let mut subscription = harness.handler.subscribe(&namespace, 10).await.unwrap();
    assert!(subscription.request_id().starts_with(REQUEST_ID_PREFIX));

    let events = collect_delivered(&mut subscription, 2).await;
    assert_eq!(events[0].request_id, subscription.request_id());
    assert_eq!(events[1].request_id, subscription.request_id());
    assert_eq!(events[0].version, 1);
    assert_eq!(events[1].version, 2);

    let first: serde_json::Value = serde_json::from_slice(&events[0].payload).unwrap();
    assert_eq!(first["n"], 1);
}

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

    harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!({"boot": true})))
        .await
        .unwrap();

    assert!(!harness.handler.is_read_stream_ready(&namespace).await);
    harness.handler.subscribe(&namespace, 1).await.unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    assert!(harness.handler.is_read_stream_ready(&namespace).await);
    assert!(harness.read_stream.stream_exists(&namespace).await.unwrap());
}

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

    harness
        .store
        .persist_next(&namespace, &payload(serde_json::json!({"phase": "history"})))
        .await
        .unwrap();

    let mut subscription = harness.handler.subscribe(&namespace, 10).await.unwrap();
    let history = collect_delivered(&mut subscription, 1).await;
    assert_eq!(history[0].version, 1);

    let live = nstreams_core::event::StreamEvent {
        namespace: namespace.as_str().to_string(),
        version: 2,
        payload: payload(serde_json::json!({"phase": "live"})),
    };
    harness.read_stream.push_live(namespace.as_str(), live).await;

    let live_delivered = collect_delivered(&mut subscription, 1).await;
    assert_eq!(live_delivered[0].version, 2);
    assert_eq!(live_delivered[0].request_id, subscription.request_id());
}

#[tokio::test]
async fn subscribe_rejects_invalid_namespace() {
    let harness = TestHarness::new();
    let namespace = StreamNamespace::from("has space");
    let result = harness.handler.subscribe(&namespace, 1).await;
    assert!(matches!(
        result,
        Err(nstreams_core::Error::InvalidNamespace(_, _))
    ));
}