liminal-server 0.4.0

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

use std::collections::BTreeSet;

use liminal_protocol::wire::{BindingEpoch, ConnectionIncarnation, Generation};

use super::dispatch_impact::{
    DispatchEffect, DispatchImpact, DispatchImpactAccumulator, DispatchTarget,
};

fn target(participant_id: u64, connection_ordinal: u64) -> DispatchTarget {
    DispatchTarget::new(
        participant_id,
        BindingEpoch::new(
            ConnectionIncarnation::new(7, connection_ordinal),
            Generation::ONE,
        ),
    )
}

#[test]
fn empty_accumulator_is_unchanged() {
    let impact = DispatchImpactAccumulator::new().finish(41);

    assert_eq!(impact, DispatchImpact::Unchanged);
    assert_eq!(impact.conversation_id(), None);
    assert!(impact.effects().is_none());
    assert!(impact.target_union().is_empty());
}

#[test]
fn truthful_effect_with_no_current_target_remains_changed() -> Result<(), String> {
    let mut accumulator = DispatchImpactAccumulator::new();
    accumulator.record(DispatchEffect::Retired, []);
    let impact = accumulator.finish(41);

    assert_eq!(impact.conversation_id(), Some(41));
    let effects = impact
        .effects()
        .ok_or_else(|| "recorded effect became unchanged".to_owned())?;
    let retired = effects
        .get(&DispatchEffect::Retired)
        .ok_or_else(|| "Retired effect was dropped".to_owned())?;
    assert!(retired.is_empty());
    assert!(impact.target_union().is_empty());
    Ok(())
}

#[test]
fn repeated_and_prefix_effects_union_without_precedence() -> Result<(), String> {
    let first = target(2, 11);
    let second = target(3, 12);
    let third = target(4, 13);
    let mut accumulator = DispatchImpactAccumulator::new();
    accumulator.record(DispatchEffect::Published, [first]);
    accumulator.record(DispatchEffect::Published, [second]);
    accumulator.record(DispatchEffect::BindingChanged, [second]);

    let mut prefix = DispatchImpactAccumulator::new();
    prefix.record(DispatchEffect::EpisodeChanged, [third]);
    prefix.record(DispatchEffect::Acknowledged, [first, third]);
    accumulator.merge(prefix);

    assert!(!accumulator.is_empty());
    let impact = accumulator.finish(82);
    let effects = impact
        .effects()
        .ok_or_else(|| "merged effects became unchanged".to_owned())?;
    assert_eq!(
        effects.get(&DispatchEffect::Published),
        Some(&BTreeSet::from([first, second]))
    );
    assert_eq!(
        effects.get(&DispatchEffect::BindingChanged),
        Some(&BTreeSet::from([second]))
    );
    assert_eq!(
        effects.get(&DispatchEffect::EpisodeChanged),
        Some(&BTreeSet::from([third]))
    );
    assert_eq!(
        effects.get(&DispatchEffect::Acknowledged),
        Some(&BTreeSet::from([first, third]))
    );
    assert_eq!(
        impact.target_union(),
        BTreeSet::from([first, second, third])
    );
    assert_eq!(first.participant_id(), 2);
    assert_eq!(
        first
            .binding_epoch()
            .connection_incarnation
            .connection_ordinal,
        11
    );
    Ok(())
}