photon_protocol/codec/protobuf/
convert.rs

1use std::time::{Duration, SystemTime, UNIX_EPOCH};
2
3use bytes::Bytes;
4
5use photon_core::types::ack::{AckResult, AckStatus};
6use photon_core::types::batch::AssembledBatch;
7use photon_core::types::id::RunId;
8use photon_core::types::metric::Metric;
9use photon_core::types::query::{
10    DataPoint, MetricQuery, MetricSeries, QueryRequest, QueryResponse, RangePoint, SeriesData,
11};
12use photon_core::types::sequence::SequenceNumber;
13
14use crate::codec::protobuf::types::{
15    MetricBatchAck, MetricBatchRequest, ProtoAckStatus, ProtoAggregatedData, ProtoDataPoint,
16    ProtoMetricQuery, ProtoMetricSeries, ProtoQueryRequest, ProtoQueryResponse, ProtoRangePoint,
17    ProtoRawData, ProtoSeriesData, WatermarkRequest, WatermarkResponse,
18};
19
20#[derive(Debug, thiserror::Error)]
21pub enum ProtoConversionError {
22    #[error("invalid run_id: {0}")]
23    InvalidRunId(String),
24
25    #[error("invalid metric key: {0}")]
26    InvalidMetricKey(String),
27
28    #[error("unrecognised ack status: {0}")]
29    UnknownAckStatus(i32),
30
31    #[error("missing required field: {0}")]
32    MissingField(&'static str),
33}
34
35impl From<&AssembledBatch> for MetricBatchRequest {
36    fn from(batch: &AssembledBatch) -> Self {
37        Self {
38            run_id: batch.run_id.to_string(),
39            sequence_number: u64::from(batch.sequence_number),
40            compressed_payload: batch.compressed_payload.to_vec(),
41            crc32: batch.crc32,
42            compressor_name: String::new(),
43            created_at_epoch_ms: system_time_to_epoch_ms(batch.created_at),
44            point_count: batch.point_count as u32,
45            uncompressed_size: batch.uncompressed_size as u32,
46        }
47    }
48}
49
50impl TryFrom<MetricBatchRequest> for AssembledBatch {
51    type Error = ProtoConversionError;
52
53    fn try_from(proto: MetricBatchRequest) -> Result<Self, Self::Error> {
54        let run_id: uuid::Uuid = proto
55            .run_id
56            .parse()
57            .map_err(|_| ProtoConversionError::InvalidRunId(proto.run_id.clone()))?;
58
59        Ok(Self {
60            run_id: RunId::from(run_id),
61            sequence_number: SequenceNumber::from(proto.sequence_number),
62            compressed_payload: Bytes::from(proto.compressed_payload),
63            crc32: proto.crc32,
64            created_at: epoch_ms_to_system_time(proto.created_at_epoch_ms),
65            point_count: proto.point_count as usize,
66            uncompressed_size: proto.uncompressed_size as usize,
67        })
68    }
69}
70
71impl From<&AckResult> for MetricBatchAck {
72    fn from(ack: &AckResult) -> Self {
73        let status = match ack.status {
74            AckStatus::Ok => ProtoAckStatus::Ok,
75            AckStatus::Duplicate => ProtoAckStatus::Duplicate,
76            AckStatus::Rejected => ProtoAckStatus::Rejected,
77        };
78
79        Self {
80            sequence_number: u64::from(ack.sequence_number),
81            status: status.into(),
82            message: String::new(),
83        }
84    }
85}
86
87impl TryFrom<MetricBatchAck> for AckResult {
88    type Error = ProtoConversionError;
89
90    fn try_from(proto: MetricBatchAck) -> Result<Self, Self::Error> {
91        let status = match ProtoAckStatus::try_from(proto.status) {
92            Ok(ProtoAckStatus::Ok) => AckStatus::Ok,
93            Ok(ProtoAckStatus::Duplicate) => AckStatus::Duplicate,
94            Ok(ProtoAckStatus::Rejected) => AckStatus::Rejected,
95            Ok(ProtoAckStatus::Unspecified) => {
96                return Err(ProtoConversionError::UnknownAckStatus(proto.status));
97            }
98            Err(_) => {
99                return Err(ProtoConversionError::UnknownAckStatus(proto.status));
100            }
101        };
102
103        Ok(Self {
104            sequence_number: SequenceNumber::from(proto.sequence_number),
105            status,
106        })
107    }
108}
109
110impl From<AckStatus> for ProtoAckStatus {
111    fn from(status: AckStatus) -> Self {
112        match status {
113            AckStatus::Ok => ProtoAckStatus::Ok,
114            AckStatus::Duplicate => ProtoAckStatus::Duplicate,
115            AckStatus::Rejected => ProtoAckStatus::Rejected,
116        }
117    }
118}
119
120impl From<&RunId> for WatermarkRequest {
121    fn from(run_id: &RunId) -> Self {
122        Self {
123            run_id: run_id.to_string(),
124        }
125    }
126}
127
128impl TryFrom<&WatermarkRequest> for RunId {
129    type Error = ProtoConversionError;
130
131    fn try_from(proto: &WatermarkRequest) -> Result<Self, Self::Error> {
132        let uuid: uuid::Uuid = proto
133            .run_id
134            .parse()
135            .map_err(|_| ProtoConversionError::InvalidRunId(proto.run_id.clone()))?;
136        Ok(RunId::from(uuid))
137    }
138}
139
140impl From<SequenceNumber> for WatermarkResponse {
141    fn from(seq: SequenceNumber) -> Self {
142        Self {
143            sequence_number: u64::from(seq),
144        }
145    }
146}
147
148impl From<WatermarkResponse> for SequenceNumber {
149    fn from(proto: WatermarkResponse) -> Self {
150        SequenceNumber::from(proto.sequence_number)
151    }
152}
153
154impl From<&QueryRequest> for ProtoQueryRequest {
155    fn from(request: &QueryRequest) -> Self {
156        Self {
157            queries: request.queries.iter().map(ProtoMetricQuery::from).collect(),
158        }
159    }
160}
161
162impl TryFrom<ProtoQueryRequest> for QueryRequest {
163    type Error = ProtoConversionError;
164
165    fn try_from(proto: ProtoQueryRequest) -> Result<Self, Self::Error> {
166        let queries = proto
167            .queries
168            .into_iter()
169            .map(MetricQuery::try_from)
170            .collect::<Result<Vec<_>, _>>()?;
171
172        Ok(Self { queries })
173    }
174}
175
176impl From<&MetricQuery> for ProtoMetricQuery {
177    fn from(query: &MetricQuery) -> Self {
178        Self {
179            run_id: query.run_id.to_string(),
180            key: query.key.as_str().to_owned(),
181            step_start: query.step_range.start,
182            step_end: query.step_range.end,
183            target_points: query.target_points as u32,
184        }
185    }
186}
187
188impl TryFrom<ProtoMetricQuery> for MetricQuery {
189    type Error = ProtoConversionError;
190
191    fn try_from(proto: ProtoMetricQuery) -> Result<Self, Self::Error> {
192        let run_id: uuid::Uuid = proto
193            .run_id
194            .parse()
195            .map_err(|_| ProtoConversionError::InvalidRunId(proto.run_id.clone()))?;
196        let key = Metric::new(&proto.key)
197            .map_err(|_| ProtoConversionError::InvalidMetricKey(proto.key.clone()))?;
198
199        Ok(Self {
200            run_id: RunId::from(run_id),
201            key,
202            step_range: proto.step_start..proto.step_end,
203            target_points: proto.target_points as usize,
204        })
205    }
206}
207
208impl From<&QueryResponse> for ProtoQueryResponse {
209    fn from(response: &QueryResponse) -> Self {
210        Self {
211            series: response
212                .series
213                .iter()
214                .map(ProtoMetricSeries::from)
215                .collect(),
216        }
217    }
218}
219
220impl TryFrom<ProtoQueryResponse> for QueryResponse {
221    type Error = ProtoConversionError;
222
223    fn try_from(proto: ProtoQueryResponse) -> Result<Self, Self::Error> {
224        let series = proto
225            .series
226            .into_iter()
227            .map(MetricSeries::try_from)
228            .collect::<Result<Vec<_>, _>>()?;
229
230        Ok(Self { series })
231    }
232}
233
234impl From<&MetricSeries> for ProtoMetricSeries {
235    fn from(series: &MetricSeries) -> Self {
236        let data = match &series.data {
237            SeriesData::Raw { points } => ProtoSeriesData::Raw(ProtoRawData {
238                points: points.iter().map(ProtoDataPoint::from).collect(),
239            }),
240            SeriesData::Aggregated { points, envelope } => {
241                ProtoSeriesData::Aggregated(ProtoAggregatedData {
242                    points: points.iter().map(ProtoDataPoint::from).collect(),
243                    envelope: envelope.iter().map(ProtoRangePoint::from).collect(),
244                })
245            }
246        };
247
248        Self {
249            run_id: series.run_id.to_string(),
250            key: series.key.as_str().to_owned(),
251            data: Some(data),
252        }
253    }
254}
255
256impl TryFrom<ProtoMetricSeries> for MetricSeries {
257    type Error = ProtoConversionError;
258
259    fn try_from(proto: ProtoMetricSeries) -> Result<Self, Self::Error> {
260        let run_id: uuid::Uuid = proto
261            .run_id
262            .parse()
263            .map_err(|_| ProtoConversionError::InvalidRunId(proto.run_id.clone()))?;
264        let key = Metric::new(&proto.key)
265            .map_err(|_| ProtoConversionError::InvalidMetricKey(proto.key.clone()))?;
266
267        let data = match proto
268            .data
269            .ok_or(ProtoConversionError::MissingField("data"))?
270        {
271            ProtoSeriesData::Raw(raw) => SeriesData::Raw {
272                points: raw.points.into_iter().map(DataPoint::from).collect(),
273            },
274            ProtoSeriesData::Aggregated(agg) => SeriesData::Aggregated {
275                points: agg.points.into_iter().map(DataPoint::from).collect(),
276                envelope: agg.envelope.into_iter().map(RangePoint::from).collect(),
277            },
278        };
279
280        Ok(Self {
281            run_id: RunId::from(run_id),
282            key,
283            data,
284        })
285    }
286}
287
288impl From<&DataPoint> for ProtoDataPoint {
289    fn from(p: &DataPoint) -> Self {
290        Self {
291            step: p.step,
292            value: p.value,
293        }
294    }
295}
296
297impl From<ProtoDataPoint> for DataPoint {
298    fn from(proto: ProtoDataPoint) -> Self {
299        Self {
300            step: proto.step,
301            value: proto.value,
302        }
303    }
304}
305
306impl From<&RangePoint> for ProtoRangePoint {
307    fn from(p: &RangePoint) -> Self {
308        Self {
309            step_start: p.step_start,
310            step_end: p.step_end,
311            min: p.min,
312            max: p.max,
313        }
314    }
315}
316
317impl From<ProtoRangePoint> for RangePoint {
318    fn from(proto: ProtoRangePoint) -> Self {
319        Self {
320            step_start: proto.step_start,
321            step_end: proto.step_end,
322            min: proto.min,
323            max: proto.max,
324        }
325    }
326}
327
328fn system_time_to_epoch_ms(time: SystemTime) -> u64 {
329    time.duration_since(UNIX_EPOCH)
330        .unwrap_or(Duration::ZERO)
331        .as_millis() as u64
332}
333
334fn epoch_ms_to_system_time(ms: u64) -> SystemTime {
335    UNIX_EPOCH + Duration::from_millis(ms)
336}
photon_protocol/codec/protobuf/convert.rs

photon_protocol/codec/protobuf/
convert.rs
