client_sync_secure/
client_sync_secure.rs

1use umadb_client::UmaDbClient;
2use umadb_dcb::{
3    DcbAppendCondition, DcbError, DcbEvent, DcbEventStoreSync, DcbQuery, DcbQueryItem, TrackingInfo,
4};
5use uuid::Uuid;
6
7fn main() -> Result<(), Box<dyn std::error::Error>> {
8    // Connect to the gRPC server
9    let url = "https://localhost:50051".to_string();
10    let client = UmaDbClient::new(url)
11        .ca_path("server.pem".to_string()) // For self-signed server certificates.
12        .api_key("umadb:example-api-key-4f7c2b1d9e5f4a038c1a".to_string())
13        .connect()?;
14
15    // Define a consistency boundary
16    let cb = DcbQuery {
17        items: vec![DcbQueryItem {
18            types: vec!["example".to_string()],
19            tags: vec!["tag1".to_string(), "tag2".to_string()],
20        }],
21    };
22
23    // Read events for a decision model
24    let mut read_response = client.read(Some(cb.clone()), None, false, None)?;
25
26    // Build decision model
27    while let Some(result) = read_response.next() {
28        match result {
29            Ok(event) => {
30                println!(
31                    "Got event at position {}: {:?}",
32                    event.position, event.event
33                );
34            }
35            Err(status) => panic!("gRPC stream error: {}", status),
36        }
37    }
38
39    // Remember the last-known position
40    let last_known_position = read_response.head().unwrap();
41    println!("Last known position is: {:?}", last_known_position);
42
43    // Produce new event, attaching some metadata (e.g. provenance) that is
44    // stored alongside the event and returned when it is read back.
45    let mut metadata = Vec::new();
46    metadata.push(("source".to_string(), "client_sync_secure".to_string()));
47    metadata.push(("correlation_id".to_string(), Uuid::new_v4().to_string()));
48    let event = DcbEvent {
49        event_type: "example".to_string(),
50        tags: vec!["tag1".to_string(), "tag2".to_string()],
51        data: b"Hello, world!".to_vec(),
52        uuid: Some(Uuid::new_v4()),
53        metadata,
54    };
55
56    // Append event in consistency boundary
57    let commit_position1 = client.append(
58        vec![event.clone()],
59        Some(DcbAppendCondition {
60            fail_if_events_match: cb.clone(),
61            after: last_known_position,
62        }),
63        None,
64    )?;
65    println!("Appended event at position: {}", commit_position1);
66
67    // Append conflicting event - expect an error
68    let conflicting_event = DcbEvent {
69        event_type: "example".to_string(),
70        tags: vec!["tag1".to_string(), "tag2".to_string()],
71        data: b"Hello, world!".to_vec(),
72        uuid: Some(Uuid::new_v4()), // different UUID
73        metadata: Vec::new(),
74    };
75    let conflicting_result = client.append(
76        vec![conflicting_event],
77        Some(DcbAppendCondition {
78            fail_if_events_match: cb.clone(),
79            after: last_known_position,
80        }),
81        None,
82    );
83
84    // Expect an integrity error
85    match conflicting_result {
86        Err(DcbError::IntegrityError(integrity_error)) => {
87            println!("Conflicting event was rejected: {:?}", integrity_error);
88        }
89        other => panic!("Expected IntegrityError, got {:?}", other),
90    }
91
92    // Conditional appends with event UUIDs are idempotent.
93    println!(
94        "Retrying to append event at position: {:?}",
95        last_known_position
96    );
97    let commit_position2 = client.append(
98        vec![event.clone()],
99        Some(DcbAppendCondition {
100            fail_if_events_match: cb.clone(),
101            after: last_known_position,
102        }),
103        None,
104    )?;
105
106    if commit_position1 == commit_position2 {
107        println!(
108            "Append method returned same commit position: {}",
109            commit_position2
110        );
111    } else {
112        panic!("Expected idempotent retry!")
113    }
114
115    // Subscribe to all events for a projection
116    let mut subscription = client.subscribe(None, None)?;
117
118    // Build an up-to-date view
119    while let Some(result) = subscription.next() {
120        match result {
121            Ok(ev) => {
122                println!("Processing event at {}: {:?}", ev.position, ev.event);
123                if ev.position == commit_position2 {
124                    println!("Projection has processed new event!");
125                    break;
126                }
127            }
128            Err(status) => panic!("gRPC stream error: {}", status),
129        }
130    }
131
132    // Track an upstream position
133    let upstream_position = client.get_tracking_info("upstream")?;
134    let next_upstream_position = upstream_position.unwrap_or(0) + 1;
135    println!("Next upstream position: {next_upstream_position}");
136    client.append(
137        vec![],
138        None,
139        Some(TrackingInfo {
140            source: "upstream".to_string(),
141            position: next_upstream_position,
142        }),
143    )?;
144    assert_eq!(
145        next_upstream_position,
146        client.get_tracking_info("upstream")?.unwrap()
147    );
148    println!("Upstream position tracked okay!");
149
150    // Try recording the same upstream position
151    let conflicting_result = client.append(
152        vec![],
153        None,
154        Some(TrackingInfo {
155            source: "upstream".to_string(),
156            position: next_upstream_position,
157        }),
158    );
159
160    // Expect an integrity error
161    match conflicting_result {
162        Err(DcbError::IntegrityError(integrity_error)) => {
163            println!(
164                "Conflicting upstream position was rejected: {:?}",
165                integrity_error
166            );
167        }
168        other => panic!("Expected IntegrityError, got {:?}", other),
169    }
170
171    Ok(())
172}
client_sync_secure/client_sync_secure.rs

client_sync_secure/
client_sync_secure.rs
