apalis_workflow/dag/
decode.rs

1use apalis_core::backend::{BackendExt, codec::Codec};
2
3/// An entry trait that handles the values of a fan-in/fanout entry
4pub trait DagCodec<B: BackendExt>: Sized {
5    /// The codec error
6    type Error;
7
8    /// Encode the input into a Vec<Compact>
9    fn encode(self) -> Result<B::Compact, Self::Error>;
10
11    /// Decode the previous input
12    fn decode(response: &B::Compact) -> Result<Self, Self::Error>;
13}
14
15impl<B, T, Err> DagCodec<B> for Vec<T>
16where
17    B: BackendExt,
18    B::Codec: Codec<T, Compact = B::Compact, Error = Err>
19        + Codec<Vec<B::Compact>, Compact = B::Compact, Error = Err>
20        + Codec<Self, Compact = B::Compact, Error = Err>,
21{
22    type Error = Err;
23    fn encode(self) -> Result<B::Compact, <B::Codec as Codec<Self>>::Error> {
24        let mut result = Vec::new();
25        for input in self {
26            result.push(B::Codec::encode(&input)?);
27        }
28        let compact = B::Codec::encode(&result)?;
29        Ok(compact)
30    }
31
32    fn decode(response: &B::Compact) -> Result<Self, Err> {
33        let decoded = B::Codec::decode(response)?;
34        Ok(decoded)
35    }
36}
37
38macro_rules! impl_entry_for_tuple {
39    // First, implement for the first type to extract the error type
40    ($T1:ident) => {
41        impl<B, Err, $T1> DagCodec<B> for ($T1,)
42        where
43            B: BackendExt,
44            B::Codec: Codec<($T1,), Compact = B::Compact, Error = Err> + Codec<Vec<B::Compact>, Compact = B::Compact, Error = Err>,
45        {
46            type Error = Err;
47
48            fn encode(self) -> Result<B::Compact, Self::Error> {
49                let result = vec![B::Codec::encode(&self)?];
50                let compact = B::Codec::encode(&result)?;
51                Ok(compact)
52            }
53            fn decode(response: &B::Compact) -> Result<Self, Err> {
54                let decoded = B::Codec::decode(response)?;
55                Ok(decoded)
56            }
57        }
58    };
59
60    // For 2+ types, use the first type's error
61    ($T1:ident, $($T:ident),+) => {
62        impl<B, Err, $T1, $($T),+> DagCodec<B> for ($T1, $($T),+)
63        where
64            B: BackendExt,
65            B::Codec: Codec<$T1, Compact = B::Compact, Error = Err> + Codec<Vec<B::Compact>, Compact = B::Compact, Error = Err>,
66            $(B::Codec: Codec<$T, Compact = B::Compact, Error = Err>,)+
67            // Ensure all errors are the same type as T1's error
68            $(
69                <B::Codec as Codec<$T>>::Error: Into<<B::Codec as Codec<$T1>>::Error>,
70            )+
71        {
72            type Error = <B::Codec as Codec<$T1>>::Error;
73
74            fn encode(self) -> Result<B::Compact, Self::Error> {
75                #[allow(non_snake_case)]
76                let ($T1, $($T),+) = self;
77
78                let mut result = vec![B::Codec::encode(&$T1)?];
79                $(
80                    result.push(B::Codec::encode(&$T).map_err(Into::into)?);
81                )+
82                let compact = B::Codec::encode(&result)?;
83                Ok(compact)
84            }
85
86            fn decode(response: &B::Compact) -> Result<Self, Err> {
87                let decoded: Vec<B::Compact> = B::Codec::decode(response)?;
88                // Count the number of types in the tuple
89                let expected_len = 1 $(+ {let _ = stringify!($T); 1})+;
90                if decoded.len() != expected_len {
91                    panic!("Expected {} elements, got {}", expected_len, decoded.len());
92                }
93
94                let mut iter = decoded.into_iter();
95
96                #[allow(non_snake_case)]
97                let $T1 = B::Codec::decode(&iter.next().unwrap())?;
98                $(
99                    #[allow(non_snake_case)]
100                    let $T = B::Codec::decode(&iter.next().unwrap()).map_err(Into::into)?;
101                )+
102
103                Ok(($T1, $($T),+))
104            }
105        }
106    };
107}
108
109impl_entry_for_tuple!(T1);
110impl_entry_for_tuple!(T1, T2);
111impl_entry_for_tuple!(T1, T2, T3);
112impl_entry_for_tuple!(T1, T2, T3, T4);
113impl_entry_for_tuple!(T1, T2, T3, T4, T5);
114impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6);
115impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7);
116impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8);
117impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8, T9);
118impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10);
119impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11);
120impl_entry_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);
121
122macro_rules! impl_entry_passthrough {
123    ($($T:ty),+) => {
124        $(
125            impl<B, Err> DagCodec<B> for $T
126            where
127                B: BackendExt,
128                B::Codec: Codec<$T, Compact = B::Compact, Error = Err>,
129            {
130                type Error = Err;
131
132                fn encode(self) -> Result<B::Compact, Self::Error> {
133                    B::Codec::encode(&self)
134                }
135
136                fn decode(response: &B::Compact) -> Result<Self, Self::Error> {
137                    B::Codec::decode(response)
138                }
139            }
140        )+
141    };
142}
143
144impl_entry_passthrough!(
145    String,
146    &'static str,
147    u8,
148    u16,
149    u32,
150    u64,
151    u128,
152    usize,
153    i8,
154    i16,
155    i32,
156    i64,
157    i128,
158    isize,
159    f32,
160    f64,
161    bool,
162    char,
163    ()
164);
165
166impl<B, T, Err> DagCodec<B> for Option<T>
167where
168    B: BackendExt,
169    B::Codec: Codec<Self, Compact = B::Compact, Error = Err>,
170{
171    type Error = Err;
172
173    fn encode(self) -> Result<B::Compact, Self::Error> {
174        B::Codec::encode(&self)
175    }
176
177    fn decode(response: &B::Compact) -> Result<Self, Self::Error> {
178        B::Codec::decode(response)
179    }
180}
181
182impl<B, T, E, Err> DagCodec<B> for Result<T, E>
183where
184    B: BackendExt,
185    B::Codec: Codec<Self, Compact = B::Compact, Error = Err>,
186{
187    type Error = Err;
188
189    fn encode(self) -> Result<B::Compact, Self::Error> {
190        B::Codec::encode(&self)
191    }
192
193    fn decode(response: &B::Compact) -> Result<Self, Self::Error> {
194        B::Codec::decode(response)
195    }
196}
apalis_workflow/dag/decode.rs

apalis_workflow/dag/
decode.rs
