Skip to main content

UmaDbClient

umadb_client

Struct UmaDbClient 

Source 
pub struct UmaDbClient { /* private fields */ }

Implementations§

Source§

impl UmaDbClient

Source

pub fn new(url: String) -> Self

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

pub fn ca_path(self, ca_path: String) -> Self

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

pub fn api_key(self, api_key: String) -> Self

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

pub fn batch_size(self, batch_size: u32) -> Self

Source

pub fn connect(&self) -> DcbResult<SyncUmaDbClient>

Examples found in repository?
examples/client_sync_insecure.rs (line 10)
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}
More examples
Hide additional examples
examples/client_sync_secure.rs (line 13)
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}
Source

pub async fn connect_async(&self) -> DcbResult<AsyncUmaDbClient>

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

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> IntoRequest<T> for T

Source§

fn into_request(self) -> Request<T>

Wrap the input message T in a tonic::Request
Source§

impl<L> LayerExt<L> for L

Source§

fn named_layer<S>(&self, service: S) -> Layered<<L as Layer<S>>::Service, S>
where L: Layer<S>,

Applies the layer to a service and wraps it in Layered.
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = Infallible

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more