distributed 4.3.0

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
146

use std::fmt;

use super::RepositoryError;

/// Stable event-stream identity used by persistence backends.
///
/// Durable stores should key aggregate streams by both aggregate type and
/// aggregate id. ID-only repository APIs remain available for the synchronous
/// in-memory path, but production adapters should prefer this type.
#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub struct StreamIdentity {
    aggregate_type: String,
    aggregate_id: String,
}

impl StreamIdentity {
    const STORAGE_KEY_DELIMITER: char = '\u{1f}';

    /// Create a validated stream identity.
    pub fn new(
        aggregate_type: impl Into<String>,
        aggregate_id: impl Into<String>,
    ) -> Result<Self, RepositoryError> {
        let aggregate_type = aggregate_type.into();
        let aggregate_id = aggregate_id.into();

        if aggregate_type.trim().is_empty() {
            return Err(RepositoryError::InvalidStreamIdentity {
                aggregate_type,
                aggregate_id,
                reason: "aggregate type must not be empty".into(),
            });
        }

        if aggregate_id.trim().is_empty() {
            return Err(RepositoryError::InvalidStreamIdentity {
                aggregate_type,
                aggregate_id,
                reason: "aggregate id must not be empty".into(),
            });
        }

        if aggregate_type.contains(Self::STORAGE_KEY_DELIMITER) {
            return Err(RepositoryError::InvalidStreamIdentity {
                aggregate_type,
                aggregate_id,
                reason: "aggregate type contains reserved delimiter".into(),
            });
        }

        if aggregate_id.contains(Self::STORAGE_KEY_DELIMITER) {
            return Err(RepositoryError::InvalidStreamIdentity {
                aggregate_type,
                aggregate_id,
                reason: "aggregate id contains reserved delimiter".into(),
            });
        }

        Ok(Self {
            aggregate_type,
            aggregate_id,
        })
    }

    /// Stable aggregate type component.
    pub fn aggregate_type(&self) -> &str {
        &self.aggregate_type
    }

    /// Aggregate id component.
    pub fn aggregate_id(&self) -> &str {
        &self.aggregate_id
    }

    pub(crate) fn storage_key(&self) -> String {
        format!(
            "{}{}{}",
            self.aggregate_type,
            Self::STORAGE_KEY_DELIMITER,
            self.aggregate_id
        )
    }
}

impl fmt::Display for StreamIdentity {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}:{}", self.aggregate_type, self.aggregate_id)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn accepts_non_empty_type_and_id() {
        let identity = StreamIdentity::new("orders", "order-1").unwrap();

        assert_eq!(identity.aggregate_type(), "orders");
        assert_eq!(identity.aggregate_id(), "order-1");
    }

    #[test]
    fn rejects_empty_aggregate_type() {
        let err = StreamIdentity::new(" ", "order-1").unwrap_err();

        assert!(matches!(
            err,
            RepositoryError::InvalidStreamIdentity { reason, .. }
                if reason == "aggregate type must not be empty"
        ));
    }

    #[test]
    fn rejects_empty_aggregate_id() {
        let err = StreamIdentity::new("orders", "").unwrap_err();

        assert!(matches!(
            err,
            RepositoryError::InvalidStreamIdentity { reason, .. }
                if reason == "aggregate id must not be empty"
        ));
    }

    #[test]
    fn rejects_reserved_delimiter_in_aggregate_type() {
        let err = StreamIdentity::new("orders\u{1f}archived", "order-1").unwrap_err();

        assert!(matches!(
            err,
            RepositoryError::InvalidStreamIdentity { reason, .. }
                if reason == "aggregate type contains reserved delimiter"
        ));
    }

    #[test]
    fn rejects_reserved_delimiter_in_aggregate_id() {
        let err = StreamIdentity::new("orders", "order\u{1f}1").unwrap_err();

        assert!(matches!(
            err,
            RepositoryError::InvalidStreamIdentity { reason, .. }
                if reason == "aggregate id contains reserved delimiter"
        ));
    }
}