massa_models/
streaming_step.rs

1use massa_serialization::{
2    Deserializer, OptionDeserializer, OptionSerializer, SerializeError, Serializer,
3    U64VarIntDeserializer, U64VarIntSerializer,
4};
5use nom::{
6    error::{context, ContextError, ParseError},
7    IResult, Parser,
8};
9use std::{marker::PhantomData, ops::Bound::Included};
10
11/// Streaming step cursor
12#[derive(PartialEq, Eq, Copy, Clone, Debug)]
13pub enum StreamingStep<T> {
14    /// Started step, only when launching the streaming
15    Started,
16    /// Ongoing step, as long as you are streaming
17    Ongoing(T),
18    /// Finished step, after all the information has been streamed
19    ///
20    /// Also can keep an indicator of the last content streamed
21    Finished(Option<T>),
22}
23
24impl<T> StreamingStep<T> {
25    /// Indicates if the current step if finished or not without caring about the values
26    pub fn finished(&self) -> bool {
27        matches!(self, StreamingStep::Finished(_))
28    }
29}
30
31/// `StreamingStep` serializer
32pub struct StreamingStepSerializer<T, ST>
33where
34    ST: Serializer<T>,
35{
36    u64_serializer: U64VarIntSerializer,
37    data_serializer: ST,
38    option_serializer: OptionSerializer<T, ST>,
39    phantom_t: PhantomData<T>,
40}
41
42impl<T, ST> StreamingStepSerializer<T, ST>
43where
44    ST: Serializer<T> + Clone,
45{
46    /// Creates a new `StreamingStep` serializer
47    pub fn new(data_serializer: ST) -> Self {
48        Self {
49            u64_serializer: U64VarIntSerializer::new(),
50            option_serializer: OptionSerializer::new(data_serializer.clone()),
51            data_serializer,
52            phantom_t: PhantomData,
53        }
54    }
55}
56
57impl<T, ST> Serializer<StreamingStep<T>> for StreamingStepSerializer<T, ST>
58where
59    ST: Serializer<T>,
60    T: std::fmt::Debug,
61{
62    fn serialize(
63        &self,
64        value: &StreamingStep<T>,
65        buffer: &mut Vec<u8>,
66    ) -> Result<(), SerializeError> {
67        match value {
68            StreamingStep::Started => self.u64_serializer.serialize(&0u64, buffer)?,
69            StreamingStep::Ongoing(data) => {
70                self.u64_serializer.serialize(&1u64, buffer)?;
71                self.data_serializer.serialize(data, buffer)?;
72            }
73            StreamingStep::Finished(opt_data) => {
74                self.u64_serializer.serialize(&2u64, buffer)?;
75                self.option_serializer.serialize(opt_data, buffer)?;
76            }
77        };
78        Ok(())
79    }
80}
81
82/// `StreamingStep` deserializer
83pub struct StreamingStepDeserializer<T, ST>
84where
85    ST: Deserializer<T>,
86    T: Clone,
87{
88    u64_deser: U64VarIntDeserializer,
89    data_deser: ST,
90    opt_deser: OptionDeserializer<T, ST>,
91    phantom_t: PhantomData<T>,
92}
93
94impl<T, ST> StreamingStepDeserializer<T, ST>
95where
96    ST: Deserializer<T> + Clone,
97    T: Clone,
98{
99    /// Creates a new `StreamingStep` deserializer
100    pub fn new(data_deser: ST) -> Self {
101        Self {
102            u64_deser: U64VarIntDeserializer::new(Included(u64::MIN), Included(u64::MAX)),
103            opt_deser: OptionDeserializer::new(data_deser.clone()),
104            data_deser,
105            phantom_t: PhantomData,
106        }
107    }
108}
109
110impl<T, ST> Deserializer<StreamingStep<T>> for StreamingStepDeserializer<T, ST>
111where
112    ST: Deserializer<T>,
113    T: Clone,
114{
115    fn deserialize<'a, E: ParseError<&'a [u8]> + ContextError<&'a [u8]>>(
116        &self,
117        buffer: &'a [u8],
118    ) -> IResult<&'a [u8], StreamingStep<T>, E> {
119        context("StreamingStep", |input| {
120            let (rest, ident) =
121                context("identifier", |input| self.u64_deser.deserialize(input)).parse(input)?;
122            match ident {
123                0u64 => Ok((rest, StreamingStep::Started)),
124                1u64 => context("ongoing data", |input| self.data_deser.deserialize(input))
125                    .map(StreamingStep::Ongoing)
126                    .parse(rest),
127                2u64 => context("finished data", |input| self.opt_deser.deserialize(input))
128                    .map(StreamingStep::Finished)
129                    .parse(rest),
130                _ => Err(nom::Err::Error(ParseError::from_error_kind(
131                    buffer,
132                    nom::error::ErrorKind::Digit,
133                ))),
134            }
135        })
136        .parse(buffer)
137    }
138}
massa_models/streaming_step.rs

massa_models/
streaming_step.rs
