liminal-server 0.20.1

Standalone server for the liminal messaging bus
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
92
93
94
95
96
97
98
99

use std::error::Error;
use std::sync::Arc;

use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore};

use super::OperationBatch;
use crate::server::participant::production::log::{OperationLog, StoredOperation};
use crate::server::participant::production::outbox_log::{OutboxLog, OutboxRow, decode_row};
use crate::server::participant::production::routed_stream::{read_head, read_page};
use crate::server::participant::production::tests_memory_store::MemoryStore;

#[test]
fn base_and_extension_stage_without_issuing_and_commit_together() -> Result<(), Box<dyn Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let shared = Arc::clone(&store) as Arc<dyn DurableStore>;
    let log = OperationLog::new(Arc::clone(&shared), 41);
    let outbox = OutboxLog::new(shared, 41);
    let mut batch = OperationBatch::new("conversation-route");
    let operation = StoredOperation::Genesis {
        event: vec![1, 2, 3],
    };
    let extension = OutboxRow::AckAdvanced {
        source_log_sequence: 0,
        participant_id: 1,
        through_seq: 9,
    };
    block_on(log.stage_into(&mut batch, &operation, 0, 17))??;
    block_on(outbox.stage_into(&mut batch, &extension, 0, 17))??;
    assert!(!log.issuance().was_submitted());
    let before = store.snapshot();
    assert_eq!(before.apply_routed_batch_calls, 0);
    assert_eq!(before.append_calls, 0);
    assert_eq!(before.append_all_calls, 0);
    assert_eq!(before.flushes, 0);
    block_on(batch.commit(store.as_ref(), || log.issuance().mark_submitted()))??;
    assert!(log.issuance().was_submitted());
    let counts = store.snapshot();
    assert_eq!(counts.apply_routed_batch_calls, 1);
    assert_eq!(counts.apply_routed_batch_items, 4);
    assert_eq!(counts.append_calls, 0);
    assert_eq!(counts.append_all_calls, 0);
    assert_eq!(counts.flushes, 0);
    let head = block_on(read_head(
        store.as_ref(),
        "conversation-route",
        "liminal:participant-production-unit2:41",
    ))??
    .ok_or("extension head missing")?;
    let rows = block_on(read_page(
        store.as_ref(),
        "conversation-route",
        "liminal:participant-production-unit2:41",
        &head,
        0,
        1,
    ))??;
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].timestamp, 17);
    assert_eq!(decode_row(&rows[0].payload)?, extension);
    Ok(())
}

#[test]
fn a_staging_refusal_does_not_mark_or_publish_the_base_row() -> Result<(), Box<dyn Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let shared = Arc::clone(&store) as Arc<dyn DurableStore>;
    let log = OperationLog::new(Arc::clone(&shared), 42);
    let outbox = OutboxLog::new(shared, 42);
    let mut batch = OperationBatch::new("conversation-route");
    block_on(log.stage_into(
        &mut batch,
        &StoredOperation::Genesis { event: vec![1] },
        0,
        20,
    ))??;
    let extension = OutboxRow::AckAdvanced {
        source_log_sequence: 0,
        participant_id: 1,
        through_seq: 9,
    };
    assert!(block_on(outbox.stage_into(&mut batch, &extension, 1, 20))?.is_err());
    assert!(!log.issuance().was_submitted());
    assert!(block_on(batch.commit(store.as_ref(), || log.issuance().mark_submitted()))?.is_err());
    assert!(!log.issuance().was_submitted());
    let counts = store.snapshot();
    assert_eq!(counts.apply_routed_batch_calls, 0);
    assert_eq!(counts.append_calls, 0);
    assert_eq!(counts.flushes, 0);
    assert!(
        block_on(read_head(
            store.as_ref(),
            "conversation-route",
            "liminal:participant-production:42"
        ))??
        .is_none()
    );
    Ok(())
}