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
use std::collections::HashMap;
use serde::{de::DeserializeOwned, Serialize};
use crate::{
models::{
records::{
CursorAndHasNext, Record, RecordWrite, RecordsRetrieveRequest, RecordsSyncRequest,
},
ItemId,
},
Items, ItemsVec, RawValue, Resource, Result,
};
pub type RecordsResource = Resource<Record<HashMap<String, RawValue>>>;
impl RecordsResource {
/// Ingest records into a stream.
///
/// Note: The maximum total request size is 10 MB.
///
/// # Arguments
///
/// * `stream_id` - ID of the stream to ingest records into.
/// * `records` - Records to ingest.
pub async fn ingest<T: Serialize>(
&self,
stream_id: &str,
records: &[RecordWrite<T>],
) -> Result<()> {
self.api_client
.post::<serde_json::Value, _>(
&format!("streams/{stream_id}/records"),
&Items::new(records),
)
.await?;
Ok(())
}
/// Upsert records into a stream.
///
/// Note: The maximum total request size is 10 MB.
///
/// # Arguments
///
/// * `stream_id` - ID of the stream to ingest records into.
/// * `records` - Records to ingest.
pub async fn upsert<T: Serialize>(
&self,
stream_id: &str,
records: &[RecordWrite<T>],
) -> Result<()> {
self.api_client
.post::<serde_json::Value, _>(
&format!("streams/{stream_id}/records/upsert"),
&Items::new(records),
)
.await?;
Ok(())
}
/// Retrieve records from a stream.
///
/// # Arguments
///
/// * `stream_id` - ID of the stream to retrieve records from.
/// * `request` - Request with optional filter and sort.
pub async fn retrieve<T: DeserializeOwned>(
&self,
stream_id: &str,
request: &RecordsRetrieveRequest,
) -> Result<ItemsVec<Record<T>>> {
self.api_client
.post(&format!("streams/{stream_id}/records/filter"), request)
.await
}
/// Subscribe to changes for records from the stream, matching a supplied filter.
///
/// # Arguments
///
/// * `stream_id` - ID of the stream to subscribe to.
/// * `request` - Request with optional filter.
pub async fn sync<T: DeserializeOwned>(
&self,
stream_id: &str,
request: &RecordsSyncRequest,
) -> Result<ItemsVec<Record<T>, CursorAndHasNext>> {
self.api_client
.post(&format!("streams/{stream_id}/records/sync"), request)
.await
}
/// Delete records from a stream.
///
/// # Arguments
///
/// * `stream_id` - ID of the stream to delete records from.
/// * `items` - IDs of the records to delete.
pub async fn delete(&self, stream_id: &str, items: &[ItemId]) -> Result<()> {
self.api_client
.post_request(&format!("streams/{stream_id}/records/delete"))
.json(&Items::new(items))?
.accept_nothing()
.send()
.await?;
Ok(())
}
}