lxmf-sdk 0.7.0

High-level Rust SDK for LXMF clients and RPC-backed LXMF workflows.
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

use super::*;

#[test]
fn propagation_local_payload_ingest_and_fetch_use_zmq_sdk_envelopes() {
    let command_endpoint = unused_loopback_endpoint();
    let response_endpoint = unused_loopback_endpoint();
    let captured = Arc::new(Mutex::new(Vec::new()));
    let server = spawn_response_sequence_zmq_server(
        command_endpoint.clone(),
        vec![
            json!({
                "response": {
                    "operation_id": "app.propagation.ingest",
                    "kind": "result",
                    "accepted": true,
                    "correlation_id": null,
                    "payload": {
                        "ingested_count": 1,
                        "duplicate_count": 0,
                        "payload_bytes": 18,
                        "transferred_bytes": 18,
                        "transient_id": "transient-sdk-ingest",
                        "propagation": {
                            "enabled": true,
                            "selected_node": "router-ingest",
                            "sync_state": 1,
                            "state_name": "queued",
                            "queue_depth": 5,
                            "total_ingested": 11,
                            "last_ingest_count": 1
                        }
                    }
                }
            }),
            json!({
                "response": {
                    "operation_id": "app.propagation.fetch",
                    "kind": "result",
                    "accepted": true,
                    "correlation_id": null,
                    "payload": {
                        "transient_id": "transient-sdk-ingest",
                        "payload_hex": "70726f7061676174696f6e2d7061796c6f6164",
                        "payload_bytes": 18,
                        "transferred_bytes": 18,
                        "propagation": {
                            "enabled": true,
                            "selected_node": "router-fetch",
                            "sync_state": 2,
                            "state_name": "serving",
                            "queue_depth": 4,
                            "client_propagation_messages_served": 1
                        }
                    }
                }
            }),
        ],
        Arc::clone(&captured),
    );
    let mut config = ZmqPipelineBackendConfig::local_tcp(command_endpoint, response_endpoint);
    config.request_timeout = std::time::Duration::from_secs(2);
    let client = ZmqPipelineBackendClient::new(config).expect("zmq client");

    let ingested = client
        .propagation_ingest(crate::PropagationIngestRequest {
            transient_id: Some("transient-sdk-ingest".to_string()),
            payload_hex: Some("70726f7061676174696f6e2d7061796c6f6164".to_string()),
        })
        .expect("propagation ingest");
    let fetched = client
        .propagation_fetch(crate::PropagationFetchRequest {
            transient_id: "transient-sdk-ingest".to_string(),
        })
        .expect("propagation fetch");

    assert_eq!(ingested.ingested_count, 1);
    assert_eq!(ingested.duplicate_count, 0);
    assert_eq!(ingested.payload_bytes, 18);
    assert_eq!(ingested.transient_id, "transient-sdk-ingest");
    assert_eq!(ingested.propagation["selected_node"], json!("router-ingest"));
    assert_eq!(ingested.recovery_state.selected_node.as_deref(), Some("router-ingest"));
    assert_eq!(ingested.recovery_state.sync_state, 1);
    assert_eq!(ingested.recovery_state.state_name.as_deref(), Some("queued"));
    assert_eq!(ingested.recovery_state.queue_depth, 5);
    assert_eq!(ingested.recovery_state.total_ingested, 11);
    assert_eq!(ingested.recovery_state.last_ingest_count, 1);
    assert_eq!(fetched.transient_id, "transient-sdk-ingest");
    assert_eq!(fetched.payload_hex, "70726f7061676174696f6e2d7061796c6f6164");
    assert_eq!(fetched.transferred_bytes, 18);
    assert_eq!(fetched.propagation["selected_node"], json!("router-fetch"));
    assert_eq!(fetched.recovery_state.selected_node.as_deref(), Some("router-fetch"));
    assert_eq!(fetched.recovery_state.sync_state, 2);
    assert_eq!(fetched.recovery_state.state_name.as_deref(), Some("serving"));
    assert_eq!(fetched.recovery_state.queue_depth, 4);
    assert_eq!(fetched.recovery_state.client_propagation_messages_served, 1);

    let captured = captured.lock().expect("captured requests");
    let operation_ids = captured
        .iter()
        .map(|request| request.params.as_ref().expect("params")["operation_id"].clone())
        .collect::<Vec<_>>();
    assert_eq!(
        operation_ids,
        vec![json!("app.propagation.ingest"), json!("app.propagation.fetch")]
    );
    let kinds = captured
        .iter()
        .map(|request| request.params.as_ref().expect("params")["kind"].clone())
        .collect::<Vec<_>>();
    assert_eq!(kinds, vec![json!("command"), json!("command")]);
    assert_eq!(
        captured[0].params.as_ref().expect("params")["payload"],
        json!({
            "transient_id": "transient-sdk-ingest",
            "payload_hex": "70726f7061676174696f6e2d7061796c6f6164"
        })
    );
    assert_eq!(
        captured[1].params.as_ref().expect("params")["payload"],
        json!({ "transient_id": "transient-sdk-ingest" })
    );
    server.join().expect("server joined");
}