s2_api/v1/stream/proto/
mod.rs

1use compact_str::ToCompactString;
2use s2_common::record;
3
4include!("s2.v1.rs");
5
6impl From<record::StreamPosition> for StreamPosition {
7    fn from(record::StreamPosition { seq_num, timestamp }: record::StreamPosition) -> Self {
8        Self { seq_num, timestamp }
9    }
10}
11
12impl From<Header> for record::Header {
13    fn from(Header { name, value }: Header) -> Self {
14        Self { name, value }
15    }
16}
17
18impl From<record::Header> for Header {
19    fn from(record::Header { name, value }: record::Header) -> Self {
20        Self { name, value }
21    }
22}
23
24impl TryFrom<AppendRecord> for s2_common::stream::AppendRecord {
25    type Error = s2_common::ValidationError;
26
27    fn try_from(
28        AppendRecord {
29            timestamp,
30            headers,
31            body,
32        }: AppendRecord,
33    ) -> Result<Self, Self::Error> {
34        Ok(Self::try_from(s2_common::stream::AppendRecordParts {
35            timestamp,
36            record: record::Record::try_from_parts(
37                headers.into_iter().map(Into::into).collect(),
38                body,
39            )
40            .map_err(|e| e.to_string())?
41            .into(),
42        })?)
43    }
44}
45
46impl TryFrom<AppendInput> for s2_common::stream::AppendInput {
47    type Error = s2_common::ValidationError;
48
49    fn try_from(
50        AppendInput {
51            records,
52            match_seq_num,
53            fencing_token,
54        }: AppendInput,
55    ) -> Result<Self, Self::Error> {
56        let records = records
57            .into_iter()
58            .map(s2_common::stream::AppendRecord::try_from)
59            .collect::<Result<Vec<_>, _>>()?;
60
61        Ok(Self {
62            records: s2_common::stream::AppendRecordBatch::try_from(records)?,
63            match_seq_num,
64            fencing_token: fencing_token
65                .as_deref()
66                .map(|s| s.to_compact_string().try_into())
67                .transpose()?,
68        })
69    }
70}
71
72impl From<s2_common::stream::AppendAck> for AppendAck {
73    fn from(
74        s2_common::stream::AppendAck { start, end, tail }: s2_common::stream::AppendAck,
75    ) -> Self {
76        Self {
77            start: Some(start.into()),
78            end: Some(end.into()),
79            tail: Some(tail.into()),
80        }
81    }
82}
83
84impl From<record::SequencedRecord> for SequencedRecord {
85    fn from(record: record::SequencedRecord) -> Self {
86        let (record::StreamPosition { seq_num, timestamp }, record) = record.into_parts();
87        let (headers, body) = record.into_parts();
88        Self {
89            seq_num,
90            timestamp,
91            headers: headers.into_iter().map(Into::into).collect(),
92            body,
93        }
94    }
95}
96
97impl From<s2_common::stream::ReadBatch> for ReadBatch {
98    fn from(batch: s2_common::stream::ReadBatch) -> Self {
99        Self {
100            records: batch.records.into_iter().map(Into::into).collect(),
101            tail: batch.tail.map(Into::into),
102        }
103    }
104}
s2_api/v1/stream/proto/mod.rs

s2_api/v1/stream/proto/
mod.rs
