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

s2_api/v1/stream/
mod.rs
