s2_api/v1/stream/
mod.rs

1#[cfg(feature = "axum")]
2pub mod extract;
3
4pub mod json;
5pub mod proto;
6pub mod s2s;
7pub mod sse;
8
9use std::time::Duration;
10
11use futures_core::stream::BoxStream;
12use itertools::Itertools as _;
13use s2_common::{
14    encryption::EncryptionKey,
15    record,
16    stream::{StreamName, StreamNamePrefix, StreamNameStartAfter},
17};
18use serde::{Deserialize, Serialize};
19use time::OffsetDateTime;
20
21use super::config::{EncryptionAlgorithm, StreamConfig};
22use crate::{data::Format, mime::JsonOrProto};
23
24#[rustfmt::skip]
25#[derive(Debug, Clone, Serialize, Deserialize)]
26#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
27pub struct StreamInfo {
28    /// Stream name.
29    pub name: StreamName,
30    /// Creation time in RFC 3339 format.
31    #[serde(with = "time::serde::rfc3339")]
32    pub created_at: OffsetDateTime,
33    /// Deletion time in RFC 3339 format, if the stream is being deleted.
34    #[serde(with = "time::serde::rfc3339::option")]
35    pub deleted_at: Option<OffsetDateTime>,
36    /// Encryption algorithm for this stream, if encryption is enabled.
37    pub cipher: Option<EncryptionAlgorithm>,
38}
39
40impl From<s2_common::stream::StreamInfo> for StreamInfo {
41    fn from(value: s2_common::stream::StreamInfo) -> Self {
42        Self {
43            name: value.name,
44            created_at: value.created_at,
45            deleted_at: value.deleted_at,
46            cipher: value.cipher.map(Into::into),
47        }
48    }
49}
50
51#[rustfmt::skip]
52#[derive(Debug, Clone, Serialize, Deserialize)]
53#[cfg_attr(feature = "utoipa", derive(utoipa::IntoParams))]
54#[cfg_attr(feature = "utoipa", into_params(parameter_in = Query))]
55pub struct ListStreamsRequest {
56    /// Filter to streams whose names begin with this prefix.
57    #[cfg_attr(feature = "utoipa", param(value_type = String, default = "", required = false))]
58    pub prefix: Option<StreamNamePrefix>,
59    /// Filter to streams whose names lexicographically start after this string.
60    #[cfg_attr(feature = "utoipa", param(value_type = String, default = "", required = false))]
61    pub start_after: Option<StreamNameStartAfter>,
62    /// Number of results, up to a maximum of 1000.
63    #[cfg_attr(feature = "utoipa", param(value_type = usize, maximum = 1000, default = 1000, required = false))]
64    pub limit: Option<usize>,
65}
66
67super::impl_list_request_conversions!(ListStreamsRequest, StreamNamePrefix, StreamNameStartAfter);
68
69#[rustfmt::skip]
70#[derive(Debug, Clone, Serialize, Deserialize)]
71#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
72pub struct ListStreamsResponse {
73    /// Matching streams.
74    #[cfg_attr(feature = "utoipa", schema(max_items = 1000))]
75    pub streams: Vec<StreamInfo>,
76    /// Indicates that there are more results that match the criteria.
77    pub has_more: bool,
78}
79
80#[rustfmt::skip]
81#[derive(Debug, Clone, Serialize, Deserialize)]
82#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
83pub struct CreateStreamRequest {
84    /// Stream name that is unique to the basin.
85    /// It can be between 1 and 512 bytes in length.
86    pub stream: StreamName,
87    /// Stream configuration.
88    pub config: Option<StreamConfig>,
89}
90
91#[rustfmt::skip]
92#[derive(Debug, Clone, Serialize, Deserialize)]
93#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
94/// Position of a record in a stream.
95pub struct StreamPosition {
96    /// Sequence number assigned by the service.
97    pub seq_num: record::SeqNum,
98    /// Timestamp, which may be client-specified or assigned by the service.
99    /// If it is assigned by the service, it will represent milliseconds since Unix epoch.
100    pub timestamp: record::Timestamp,
101}
102
103impl From<record::StreamPosition> for StreamPosition {
104    fn from(pos: record::StreamPosition) -> Self {
105        Self {
106            seq_num: pos.seq_num,
107            timestamp: pos.timestamp,
108        }
109    }
110}
111
112impl From<StreamPosition> for record::StreamPosition {
113    fn from(pos: StreamPosition) -> Self {
114        Self {
115            seq_num: pos.seq_num,
116            timestamp: pos.timestamp,
117        }
118    }
119}
120
121#[rustfmt::skip]
122#[derive(Debug, Clone, Serialize, Deserialize)]
123#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
124pub struct TailResponse {
125    /// Sequence number that will be assigned to the next record on the stream, and timestamp of the last record.
126    pub tail: StreamPosition,
127}
128
129#[rustfmt::skip]
130#[derive(Debug, Clone, Serialize, Deserialize)]
131#[cfg_attr(feature = "utoipa", derive(utoipa::IntoParams))]
132#[cfg_attr(feature = "utoipa", into_params(parameter_in = Query))]
133pub struct ReadStart {
134    /// Start from a sequence number.
135    #[cfg_attr(feature = "utoipa", param(value_type = record::SeqNum, required = false))]
136    pub seq_num: Option<record::SeqNum>,
137    /// Start from a timestamp.
138    #[cfg_attr(feature = "utoipa", param(value_type = record::Timestamp, required = false))]
139    pub timestamp: Option<record::Timestamp>,
140    /// Start from number of records before the next sequence number.
141    #[cfg_attr(feature = "utoipa", param(value_type = u64, required = false))]
142    pub tail_offset: Option<u64>,
143    /// Start reading from the tail if the requested position is beyond it.
144    /// Otherwise, a `416 Range Not Satisfiable` response is returned.
145    #[cfg_attr(feature = "utoipa", param(value_type = bool, required = false))]
146    pub clamp: Option<bool>,
147}
148
149impl TryFrom<ReadStart> for s2_common::stream::ReadStart {
150    type Error = s2_common::ValidationError;
151
152    fn try_from(value: ReadStart) -> Result<Self, Self::Error> {
153        let from = match (value.seq_num, value.timestamp, value.tail_offset) {
154            (Some(seq_num), None, None) => s2_common::stream::ReadFrom::SeqNum(seq_num),
155            (None, Some(timestamp), None) => s2_common::stream::ReadFrom::Timestamp(timestamp),
156            (None, None, Some(tail_offset)) => s2_common::stream::ReadFrom::TailOffset(tail_offset),
157            (None, None, None) => s2_common::stream::ReadFrom::TailOffset(0),
158            _ => {
159                return Err(s2_common::ValidationError(
160                    "only one of seq_num, timestamp, or tail_offset can be provided".to_owned(),
161                ));
162            }
163        };
164        let clamp = value.clamp.unwrap_or(false);
165        Ok(Self { from, clamp })
166    }
167}
168
169#[rustfmt::skip]
170#[derive(Debug, Clone, Serialize, Deserialize)]
171#[cfg_attr(feature = "utoipa", derive(utoipa::IntoParams))]
172#[cfg_attr(feature = "utoipa", into_params(parameter_in = Query))]
173pub struct ReadEnd {
174    /// Record count limit.
175    /// Non-streaming reads are capped by the default limit of 1000 records.
176    #[cfg_attr(feature = "utoipa", param(value_type = u64, required = false))]
177    pub count: Option<usize>,
178    /// Metered bytes limit.
179    /// Non-streaming reads are capped by the default limit of 1 MiB.
180    #[cfg_attr(feature = "utoipa", param(value_type = usize, required = false))]
181    pub bytes: Option<usize>,
182    /// Exclusive timestamp to read until.
183    #[cfg_attr(feature = "utoipa", param(value_type = record::Timestamp, required = false))]
184    pub until: Option<record::Timestamp>,
185    /// Duration in seconds to wait for new records.
186    /// The default duration is 0 if there is a bound on `count`, `bytes`, or `until`, and otherwise infinite.
187    /// Non-streaming reads are always bounded on `count` and `bytes`, so you can achieve long poll semantics by specifying a non-zero duration up to 60 seconds.
188    /// In the context of an SSE or S2S streaming read, the duration will bound how much time can elapse between records throughout the lifetime of the session.
189    #[cfg_attr(feature = "utoipa", param(value_type = u32, required = false))]
190    pub wait: Option<u32>,
191}
192
193impl From<ReadEnd> for s2_common::stream::ReadEnd {
194    fn from(value: ReadEnd) -> Self {
195        Self {
196            limit: s2_common::read_extent::ReadLimit::from_count_and_bytes(
197                value.count,
198                value.bytes,
199            ),
200            until: value.until.into(),
201            wait: value.wait.map(|w| Duration::from_secs(w as u64)),
202        }
203    }
204}
205
206#[derive(Debug, Clone)]
207pub enum ReadRequest {
208    /// Unary
209    Unary {
210        encryption_key: Option<EncryptionKey>,
211        format: Format,
212        response_mime: JsonOrProto,
213    },
214    /// Server-Sent Events streaming response
215    EventStream {
216        encryption_key: Option<EncryptionKey>,
217        format: Format,
218        last_event_id: Option<sse::LastEventId>,
219    },
220    /// S2S streaming response
221    S2s {
222        encryption_key: Option<EncryptionKey>,
223        response_compression: s2s::CompressionAlgorithm,
224    },
225}
226
227pub enum AppendRequest {
228    /// Unary
229    Unary {
230        encryption_key: Option<EncryptionKey>,
231        input: s2_common::stream::AppendInput,
232        response_mime: JsonOrProto,
233    },
234    /// S2S bi-directional streaming
235    S2s {
236        encryption_key: Option<EncryptionKey>,
237        inputs: BoxStream<'static, Result<s2_common::stream::AppendInput, AppendInputStreamError>>,
238        response_compression: s2s::CompressionAlgorithm,
239    },
240}
241
242impl std::fmt::Debug for AppendRequest {
243    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
244        match self {
245            AppendRequest::Unary {
246                encryption_key,
247                input,
248                response_mime: response,
249            } => f
250                .debug_struct("AppendRequest::Unary")
251                .field("encryption_key", encryption_key)
252                .field("input", input)
253                .field("response", response)
254                .finish(),
255            AppendRequest::S2s {
256                encryption_key,
257                response_compression,
258                ..
259            } => f
260                .debug_struct("AppendRequest::S2s")
261                .field("encryption_key", encryption_key)
262                .field("response_compression", response_compression)
263                .finish(),
264        }
265    }
266}
267
268#[derive(Debug, thiserror::Error)]
269pub enum AppendInputStreamError {
270    #[error("Failed to decode S2S frame: {0}")]
271    FrameDecode(#[from] std::io::Error),
272    #[error(transparent)]
273    Validation(#[from] s2_common::ValidationError),
274}
275
276#[derive(Debug, Clone, Serialize, Deserialize)]
277#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
278pub struct Header(pub String, pub String);
279
280#[rustfmt::skip]
281/// Record that is durably sequenced on a stream.
282#[derive(Debug, Clone, Serialize)]
283#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
284pub struct SequencedRecord {
285    /// Sequence number assigned by the service.
286    pub seq_num: record::SeqNum,
287    /// Timestamp for this record.
288    pub timestamp: record::Timestamp,
289    /// Series of name-value pairs for this record.
290    #[serde(default, skip_serializing_if = "Vec::is_empty")]
291    #[cfg_attr(feature = "utoipa", schema(required = false))]
292    pub headers: Vec<Header>,
293    /// Body of the record.
294    #[serde(default, skip_serializing_if = "String::is_empty")]
295    #[cfg_attr(feature = "utoipa", schema(required = false))]
296    pub body: String,
297}
298
299impl SequencedRecord {
300    pub fn encode(format: Format, record: record::SequencedRecord) -> Self {
301        let (record::StreamPosition { seq_num, timestamp }, record) = record.into_parts();
302        let (headers, body) = record.into_parts();
303        Self {
304            seq_num,
305            timestamp,
306            headers: headers
307                .into_iter()
308                .map(|h| Header(format.encode(&h.name), format.encode(&h.value)))
309                .collect(),
310            body: format.encode(&body),
311        }
312    }
313}
314
315#[rustfmt::skip]
316/// Record to be appended to a stream.
317#[derive(Debug, Clone, Serialize, Deserialize)]
318#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
319pub struct AppendRecord {
320    /// Timestamp for this record.
321    /// The service will always ensure monotonicity by adjusting it up if necessary to the maximum observed timestamp.
322    /// Refer to stream timestamping configuration for the finer semantics around whether a client-specified timestamp is required, and whether it will be capped at the arrival time.
323    pub timestamp: Option<record::Timestamp>,
324    /// Series of name-value pairs for this record.
325    #[serde(default, skip_serializing_if = "Vec::is_empty")]
326    #[cfg_attr(feature = "utoipa", schema(required = false))]
327    pub headers: Vec<Header>,
328    /// Body of the record.
329    #[serde(default, skip_serializing_if = "String::is_empty")]
330    #[cfg_attr(feature = "utoipa", schema(required = false))]
331    pub body: String,
332}
333
334impl AppendRecord {
335    pub fn decode(
336        self,
337        format: Format,
338    ) -> Result<s2_common::stream::AppendRecord, s2_common::ValidationError> {
339        let headers = self
340            .headers
341            .into_iter()
342            .map(|Header(name, value)| {
343                Ok::<record::Header, s2_common::ValidationError>(record::Header {
344                    name: format.decode(name)?,
345                    value: format.decode(value)?,
346                })
347            })
348            .try_collect()?;
349
350        let body = format.decode(self.body)?;
351
352        let record = record::Record::try_from_parts(headers, body)
353            .map_err(|e| e.to_string())?
354            .into();
355
356        let parts = s2_common::stream::AppendRecordParts {
357            timestamp: self.timestamp,
358            record,
359        };
360
361        s2_common::stream::AppendRecord::try_from(parts)
362            .map_err(|e| s2_common::ValidationError(e.to_string()))
363    }
364}
365
366#[rustfmt::skip]
367/// Payload of an `append` request.
368#[derive(Debug, Clone, Serialize, Deserialize)]
369#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
370pub struct AppendInput {
371    /// Batch of records to append atomically, which must contain at least one record, and no more than 1000.
372    /// The total size of a batch of records may not exceed 1 MiB of metered bytes.
373    pub records: Vec<AppendRecord>,
374    /// Enforce that the sequence number assigned to the first record matches.
375    pub match_seq_num: Option<record::SeqNum>,
376    /// Enforce a fencing token, which starts out as an empty string that can be overridden by a `fence` command record.
377    pub fencing_token: Option<record::FencingToken>,
378}
379
380impl AppendInput {
381    pub fn decode(
382        self,
383        format: Format,
384    ) -> Result<s2_common::stream::AppendInput, s2_common::ValidationError> {
385        let records: Vec<s2_common::stream::AppendRecord> = self
386            .records
387            .into_iter()
388            .map(|record| record.decode(format))
389            .try_collect()?;
390
391        Ok(s2_common::stream::AppendInput {
392            records: s2_common::stream::AppendRecordBatch::try_from(records)?,
393            match_seq_num: self.match_seq_num,
394            fencing_token: self.fencing_token,
395        })
396    }
397}
398
399#[rustfmt::skip]
400/// Success response to an `append` request.
401#[derive(Debug, Clone, Serialize)]
402#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
403pub struct AppendAck {
404    /// Sequence number and timestamp of the first record that was appended.
405    pub start: StreamPosition,
406    /// Sequence number of the last record that was appended `+ 1`, and timestamp of the last record that was appended.
407    /// The difference between `end.seq_num` and `start.seq_num` will be the number of records appended.
408    pub end: StreamPosition,
409    /// Sequence number that will be assigned to the next record on the stream, and timestamp of the last record on the stream.
410    /// This can be greater than the `end` position in case of concurrent appends.
411    pub tail: StreamPosition,
412}
413
414impl From<s2_common::stream::AppendAck> for AppendAck {
415    fn from(ack: s2_common::stream::AppendAck) -> Self {
416        Self {
417            start: ack.start.into(),
418            end: ack.end.into(),
419            tail: ack.tail.into(),
420        }
421    }
422}
423
424#[rustfmt::skip]
425/// Aborted due to a failed condition.
426#[derive(Debug, Clone, Serialize, Deserialize)]
427#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
428#[serde(rename_all = "snake_case")]
429pub enum AppendConditionFailed {
430    /// Fencing token did not match.
431    /// The expected fencing token is returned.
432    #[cfg_attr(feature = "utoipa", schema(title = "fencing token"))]
433    FencingTokenMismatch(record::FencingToken),
434    /// Sequence number did not match the tail of the stream.
435    /// The expected next sequence number is returned.
436    #[cfg_attr(feature = "utoipa", schema(title = "seq num"))]
437    SeqNumMismatch(record::SeqNum),
438}
439
440#[rustfmt::skip]
441#[derive(Debug, Clone, Serialize)]
442#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
443pub struct ReadBatch {
444    /// Records that are durably sequenced on the stream, retrieved based on the requested criteria.
445    /// This can only be empty in response to a unary read (i.e. not SSE), if the request cannot be satisfied without violating an explicit bound (`count`, `bytes`, or `until`).
446    pub records: Vec<SequencedRecord>,
447    /// Sequence number that will be assigned to the next record on the stream, and timestamp of the last record.
448    /// This will only be present when reading recent records.
449    #[serde(skip_serializing_if = "Option::is_none")]
450    pub tail: Option<StreamPosition>,
451}
452
453impl ReadBatch {
454    pub fn encode(format: Format, batch: s2_common::stream::ReadBatch) -> Self {
455        Self {
456            records: batch
457                .records
458                .into_iter()
459                .map(|record| SequencedRecord::encode(format, record))
460                .collect(),
461            tail: batch.tail.map(Into::into),
462        }
463    }
464}
s2_api/v1/stream/mod.rs

s2_api/v1/stream/
mod.rs
