client_sync_insecure/
client_sync_insecure.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 = "http://localhost:50051".to_string();
10    let client = UmaDbClient::new(url).connect()?;
11
12    // Define a consistency boundary
13    let boundary = DcbQuery::new().item(
14        DcbQueryItem::new()
15            .types(["example"])
16            .tags(["tag1", "tag2"]),
17    );
18
19    // Read events for a decision model
20    let mut read_response = client.read(Some(boundary.clone()), None, false, None)?;
21
22    // Build decision model
23    while let Some(result) = read_response.next() {
24        match result {
25            Ok(event) => {
26                println!(
27                    "Got event at position {}: {:?}",
28                    event.position, event.event
29                );
30            }
31            Err(status) => panic!("gRPC stream error: {}", status),
32        }
33    }
34
35    // Remember the last-known position
36    let last_known_position = read_response.head().unwrap();
37    println!("Last known position is: {:?}", last_known_position);
38
39    // Produce new event, attaching some metadata (e.g. provenance) that is
40    // stored alongside the event and returned when it is read back.
41    let event = DcbEvent::default()
42        .event_type("example")
43        .tags(["tag1", "tag2"])
44        .data(b"Hello, world!")
45        .uuid(Uuid::new_v4())
46        .metadata_entry("source", "client_sync_insecure")
47        .metadata_entry("correlation_id", Uuid::new_v4().to_string());
48
49    // Append event in consistency boundary
50    let append_condition = DcbAppendCondition {
51        fail_if_events_match: boundary.clone(),
52        after: last_known_position,
53    };
54    let position1 = client.append(vec![event.clone()], Some(append_condition.clone()), None)?;
55
56    println!("Appended event at position: {}", position1);
57
58    // Append conflicting event - expect an error
59    let conflicting_event = DcbEvent::default()
60        .event_type("example")
61        .tags(["tag1", "tag2"])
62        .data(b"Hello, world!")
63        .uuid(Uuid::new_v4()); // different UUID
64
65    let conflicting_result = client.append(
66        vec![conflicting_event],
67        Some(append_condition.clone()),
68        None,
69    );
70
71    // Expect an integrity error
72    match conflicting_result {
73        Err(DcbError::IntegrityError(integrity_error)) => {
74            println!("Conflicting event was rejected: {:?}", integrity_error);
75        }
76        other => panic!("Expected IntegrityError, got {:?}", other),
77    }
78
79    // Appending with identical event IDs and append condition is idempotent.
80    println!(
81        "Retrying to append event at position: {:?}",
82        last_known_position
83    );
84    let position2 = client.append(vec![event.clone()], Some(append_condition.clone()), None)?;
85
86    if position1 == position2 {
87        println!("Append method returned same commit position: {}", position2);
88    } else {
89        panic!("Expected idempotent retry!")
90    }
91
92    // Subscribe to all events for a projection
93    let mut subscription = client.subscribe(None, None)?;
94
95    // Build an up-to-date view
96    while let Some(result) = subscription.next() {
97        match result {
98            Ok(ev) => {
99                println!("Processing event at {}: {:?}", ev.position, ev.event);
100                if ev.position == position2 {
101                    println!("Projection has processed new event!");
102                    break;
103                }
104            }
105            Err(status) => panic!("gRPC stream error: {}", status),
106        }
107    }
108
109    // Track an upstream position
110    let upstream_position = client.get_tracking_info("upstream")?;
111    let next_upstream_position = upstream_position.unwrap_or(0) + 1;
112    println!("Next upstream position: {next_upstream_position}");
113    client.append(
114        vec![],
115        None,
116        Some(TrackingInfo {
117            source: "upstream".to_string(),
118            position: next_upstream_position,
119        }),
120    )?;
121    assert_eq!(
122        next_upstream_position,
123        client.get_tracking_info("upstream")?.unwrap()
124    );
125    println!("Upstream position tracked okay!");
126
127    // Try recording the same upstream position
128    let conflicting_result = client.append(
129        vec![],
130        None,
131        Some(TrackingInfo {
132            source: "upstream".to_string(),
133            position: next_upstream_position,
134        }),
135    );
136
137    // Expect an integrity error
138    match conflicting_result {
139        Err(DcbError::IntegrityError(integrity_error)) => {
140            println!(
141                "Conflicting upstream position was rejected: {:?}",
142                integrity_error
143            );
144        }
145        other => panic!("Expected IntegrityError, got {:?}", other),
146    }
147
148    Ok(())
149}
client_sync_insecure/client_sync_insecure.rs

client_sync_insecure/
client_sync_insecure.rs
