distributed 4.4.1

CQRS/ES framework for Rust using Plain Old Rust Structs — append-only events, replay, snapshots, outbox, service bus, and pluggable infrastructure
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! Basic microsvc integration tests — exercises dispatch with a real repository.

use distributed::microsvc::{Context, HandlerError, Routes, Service, Session};
use distributed::{AggregateBuilder, InMemoryRepository};
use serde_json::json;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use crate::models::counter::{Counter, CreateCounter, DecrementCounter, IncrementCounter};

#[tokio::test]
async fn full_lifecycle() {
    let repo = InMemoryRepository::new();
    let counter_repo = repo.clone().aggregate::<Counter>();
    let service = Service::new().routes(
        Routes::new()
            .with_dependencies(repo)
            .command("counter.initialize")
            .handle(|ctx: &Context<InMemoryRepository>| {
                let input = ctx.input::<CreateCounter>();
                let counter_repo = ctx.repo().clone().aggregate::<Counter>();
                async move {
                    let input = input?;
                    let mut counter = Counter::default();
                    counter.create(input.id.clone())?;
                    counter_repo.commit(&mut counter).await?;
                    Ok(json!({ "id": input.id }))
                }
            })
            .command("counter.increment")
            .handle(|ctx: &Context<InMemoryRepository>| {
                let input = ctx.input::<IncrementCounter>();
                let counter_repo = ctx.repo().clone().aggregate::<Counter>();
                async move {
                    let input = input?;
                    let mut counter: Counter = counter_repo
                        .get(&input.id)
                        .await?
                        .ok_or_else(|| HandlerError::NotFound(input.id.clone()))?;
                    counter.increment(input.amount)?;
                    counter_repo.commit(&mut counter).await?;
                    Ok(json!({ "value": counter.value }))
                }
            })
            .command("counter.decrement")
            .handle(|ctx: &Context<InMemoryRepository>| {
                let input = ctx.input::<DecrementCounter>();
                let counter_repo = ctx.repo().clone().aggregate::<Counter>();
                async move {
                    let input = input?;
                    let mut counter: Counter = counter_repo
                        .get(&input.id)
                        .await?
                        .ok_or_else(|| HandlerError::NotFound(input.id.clone()))?;
                    counter.decrement(input.amount)?;
                    counter_repo.commit(&mut counter).await?;
                    Ok(json!({ "value": counter.value }))
                }
            }),
    );

    // Create
    let result = service
        .dispatch("counter.initialize", json!({ "id": "c1" }), Session::new())
        .await
        .unwrap();
    assert_eq!(result, json!({ "id": "c1" }));

    // Increment twice
    let result = service
        .dispatch(
            "counter.increment",
            json!({ "id": "c1", "amount": 5 }),
            Session::new(),
        )
        .await
        .unwrap();
    assert_eq!(result, json!({ "value": 5 }));

    service
        .dispatch(
            "counter.increment",
            json!({ "id": "c1", "amount": 3 }),
            Session::new(),
        )
        .await
        .unwrap();

    // Decrement
    let result = service
        .dispatch(
            "counter.decrement",
            json!({ "id": "c1", "amount": 2 }),
            Session::new(),
        )
        .await
        .unwrap();
    assert_eq!(result, json!({ "value": 6 }));

    // Verify final state via repo
    let counter: Counter = counter_repo.get("c1").await.unwrap().unwrap();
    assert_eq!(counter.value, 6);
}

#[tokio::test]
async fn domain_event_fans_out_to_multiple_subscribers() {
    let projections = Arc::new(AtomicUsize::new(0));
    let policies = Arc::new(AtomicUsize::new(0));
    let projection_handler_count = projections.clone();
    let policy_handler_count = policies.clone();
    let service = Service::new()
        .routes(
            Routes::new()
                .with_dependencies(projections.clone())
                .event("metachangeset.opened")
                .handle(move |_ctx: &Context<Arc<AtomicUsize>>| {
                    let count = projection_handler_count.clone();
                    async move {
                        count.fetch_add(1, Ordering::SeqCst);
                        Ok(json!({ "subscriber": "projection" }))
                    }
                }),
        )
        .routes(
            Routes::new()
                .with_dependencies(policies.clone())
                .event("metachangeset.opened")
                .handle(move |_ctx: &Context<Arc<AtomicUsize>>| {
                    let count = policy_handler_count.clone();
                    async move {
                        count.fetch_add(1, Ordering::SeqCst);
                        Ok(json!({ "subscriber": "policy" }))
                    }
                }),
        );

    let message = distributed::microsvc::Message::new(
        "metachangeset.opened",
        distributed::microsvc::MessageKind::Event,
        b"{}".to_vec(),
    );
    service.dispatch_message(&message).await.unwrap();
    assert_eq!(projections.load(Ordering::SeqCst), 1);
    assert_eq!(policies.load(Ordering::SeqCst), 1);
}