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

s2_api/v1/stream/
mod.rs
