client_async_insecure/
client_async_insecure.rs

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

client_async_insecure/
client_async_insecure.rs
