liminal-server 0.25.4

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

use super::{TestResult, config_with, stop_server};
use crate::server::connection::services::open_persistent_store;
use crate::server::connection::{ConnectionSupervisor, LiminalConnectionServices};
use crate::server::runtime::start_self_stream_publisher;
use crate::server::shutdown::ShutdownHandle;
use liminal::durability::bridge::block_on;
use std::sync::Arc;

#[test]
fn fixture_stop_finalizes_before_retiring_records_and_reopens_its_store() -> TestResult {
    let store = tempfile::tempdir()?;
    let config = config_with(store.path(), 0, &[])?;
    let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
    services.signals().wake().hold();
    let counts = services.store_counts();
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    let mut publisher = start_self_stream_publisher(&services, &supervisor, &config, &shutdown)?
        .ok_or("self-stream publisher missing")?;
    let streams = services
        .signals()
        .streams()
        .cloned()
        .ok_or("streams missing")?;
    let mut accepts_stopped = 0;
    let stopped = stop_server(
        &supervisor,
        || {
            accepts_stopped += 1;
            Ok(())
        },
        &mut publisher,
    );
    let after_stop = counts.snapshot();
    let repeated = publisher.finish();
    let completions = publisher.completion_counts();
    assert_eq!(accepts_stopped, 1);
    assert_eq!(
        counts.snapshot(),
        after_stop,
        "repeat finish wrote after shutdown"
    );
    drop((streams, publisher, supervisor, services));
    stopped.map_err(|error| format!("fixture production stop: {error}"))?;
    repeated?;
    assert_eq!(completions, (1, 0, 0));
    let reopened = open_persistent_store(store.path())?;
    assert!(block_on(reopened.read_at("liminal.cost:0", 0))??.is_some());
    drop(reopened);
    store.close()?;
    Ok(())
}

#[test]
fn cleanly_stopped_drainer_is_named_stopped_when_started_again() -> TestResult {
    let store = tempfile::tempdir()?;
    let config = config_with(store.path(), 0, &[])?;
    let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
    services.signals().wake().hold();
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    let mut publisher = start_self_stream_publisher(&services, &supervisor, &config, &shutdown)?
        .ok_or("self-stream publisher missing")?;
    let streams = services
        .signals()
        .streams()
        .cloned()
        .ok_or("streams missing")?;
    stop_server(&supervisor, || Ok(()), &mut publisher)?;
    let error = streams.start().err().ok_or("stopped drainer restarted")?;
    drop((streams, publisher, supervisor, services));
    store.close()?;
    assert_eq!(
        error.to_string(),
        "listener accept failed: SELF_STREAM_EVENTS_APPEND_FAILED: record drainer is stopped"
    );
    Ok(())
}