temporalio_common/
data_converters.rs

1//! Contains traits for and default implementations of data converters, codecs, and other
2//! serialization related functionality.
3
4use crate::protos::temporal::api::{common::v1::Payload, failure::v1::Failure};
5use futures::{FutureExt, future::BoxFuture};
6use std::{collections::HashMap, sync::Arc};
7
8/// Combines a [`PayloadConverter`], [`FailureConverter`], and [`PayloadCodec`] to handle all
9/// serialization needs for communicating with the Temporal server.
10#[derive(Clone)]
11pub struct DataConverter {
12    payload_converter: PayloadConverter,
13    #[allow(dead_code)] // Will be used for failure conversion
14    failure_converter: Arc<dyn FailureConverter + Send + Sync>,
15    codec: Arc<dyn PayloadCodec + Send + Sync>,
16}
17
18impl std::fmt::Debug for DataConverter {
19    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20        f.debug_struct("DataConverter")
21            .field("payload_converter", &self.payload_converter)
22            .finish_non_exhaustive()
23    }
24}
25impl DataConverter {
26    /// Create a new DataConverter with the given payload converter, failure converter, and codec.
27    pub fn new(
28        payload_converter: PayloadConverter,
29        failure_converter: impl FailureConverter + Send + Sync + 'static,
30        codec: impl PayloadCodec + Send + Sync + 'static,
31    ) -> Self {
32        Self {
33            payload_converter,
34            failure_converter: Arc::new(failure_converter),
35            codec: Arc::new(codec),
36        }
37    }
38
39    /// Serialize a value into a single payload, applying the codec.
40    pub async fn to_payload<T: TemporalSerializable + 'static>(
41        &self,
42        data: &SerializationContextData,
43        val: &T,
44    ) -> Result<Payload, PayloadConversionError> {
45        let context = SerializationContext {
46            data,
47            converter: &self.payload_converter,
48        };
49        let payload = self.payload_converter.to_payload(&context, val)?;
50        let encoded = self.codec.encode(data, vec![payload]).await;
51        encoded
52            .into_iter()
53            .next()
54            .ok_or(PayloadConversionError::WrongEncoding)
55    }
56
57    /// Deserialize a value from a single payload, applying the codec.
58    pub async fn from_payload<T: TemporalDeserializable + 'static>(
59        &self,
60        data: &SerializationContextData,
61        payload: Payload,
62    ) -> Result<T, PayloadConversionError> {
63        let context = SerializationContext {
64            data,
65            converter: &self.payload_converter,
66        };
67        let decoded = self.codec.decode(data, vec![payload]).await;
68        let payload = decoded
69            .into_iter()
70            .next()
71            .ok_or(PayloadConversionError::WrongEncoding)?;
72        self.payload_converter.from_payload(&context, payload)
73    }
74
75    /// Serialize a value into multiple payloads (e.g. for multi-arg support), applying the codec.
76    pub async fn to_payloads<T: TemporalSerializable + 'static>(
77        &self,
78        data: &SerializationContextData,
79        val: &T,
80    ) -> Result<Vec<Payload>, PayloadConversionError> {
81        let context = SerializationContext {
82            data,
83            converter: &self.payload_converter,
84        };
85        let payloads = self.payload_converter.to_payloads(&context, val)?;
86        Ok(self.codec.encode(data, payloads).await)
87    }
88
89    /// Deserialize a value from multiple payloads (e.g. for multi-arg support), applying the codec.
90    pub async fn from_payloads<T: TemporalDeserializable + 'static>(
91        &self,
92        data: &SerializationContextData,
93        payloads: Vec<Payload>,
94    ) -> Result<T, PayloadConversionError> {
95        let context = SerializationContext {
96            data,
97            converter: &self.payload_converter,
98        };
99        let decoded = self.codec.decode(data, payloads).await;
100        self.payload_converter.from_payloads(&context, decoded)
101    }
102
103    /// Returns the payload converter component of this data converter.
104    pub fn payload_converter(&self) -> &PayloadConverter {
105        &self.payload_converter
106    }
107
108    /// Returns the codec component of this data converter.
109    pub fn codec(&self) -> &(dyn PayloadCodec + Send + Sync) {
110        self.codec.as_ref()
111    }
112}
113
114/// Data about the serialization context, indicating where the serialization is occurring.
115#[derive(Clone, Copy, Debug, PartialEq, Eq)]
116pub enum SerializationContextData {
117    /// Serialization is occurring in a workflow context.
118    Workflow,
119    /// Serialization is occurring in an activity context.
120    Activity,
121    /// Serialization is occurring in a nexus context.
122    Nexus,
123    /// No specific serialization context.
124    None,
125}
126
127/// Context for serialization operations, including the kind of context and the
128/// payload converter for nested serialization.
129#[derive(Clone, Copy)]
130pub struct SerializationContext<'a> {
131    /// The kind of serialization context (workflow, activity, etc.).
132    pub data: &'a SerializationContextData,
133    /// Allows nested types to serialize their contents using the same converter.
134    pub converter: &'a PayloadConverter,
135}
136/// Converts values to and from [`Payload`]s using different encoding strategies.
137#[derive(Clone)]
138pub enum PayloadConverter {
139    /// Uses a serde-based converter for encoding/decoding.
140    Serde(Arc<dyn ErasedSerdePayloadConverter>),
141    /// This variant signals the user wants to delegate to wrapper types
142    UseWrappers,
143    /// Tries multiple converters in order until one succeeds.
144    Composite(Arc<CompositePayloadConverter>),
145}
146
147impl std::fmt::Debug for PayloadConverter {
148    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
149        match self {
150            PayloadConverter::Serde(_) => write!(f, "PayloadConverter::Serde(...)"),
151            PayloadConverter::UseWrappers => write!(f, "PayloadConverter::UseWrappers"),
152            PayloadConverter::Composite(_) => write!(f, "PayloadConverter::Composite(...)"),
153        }
154    }
155}
156impl PayloadConverter {
157    /// Create a payload converter that uses JSON serialization via serde.
158    pub fn serde_json() -> Self {
159        Self::Serde(Arc::new(SerdeJsonPayloadConverter))
160    }
161    // TODO [rust-sdk-branch]: Proto binary, other standard built-ins
162}
163
164impl Default for PayloadConverter {
165    fn default() -> Self {
166        Self::Composite(Arc::new(CompositePayloadConverter {
167            converters: vec![Self::UseWrappers, Self::serde_json()],
168        }))
169    }
170}
171
172/// Errors that can occur during payload conversion.
173#[derive(Debug)]
174pub enum PayloadConversionError {
175    /// The payload's encoding does not match what the converter expects.
176    WrongEncoding,
177    /// An error occurred during encoding or decoding.
178    EncodingError(Box<dyn std::error::Error + Send + Sync>),
179}
180
181impl std::fmt::Display for PayloadConversionError {
182    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
183        match self {
184            PayloadConversionError::WrongEncoding => write!(f, "Wrong encoding"),
185            PayloadConversionError::EncodingError(err) => write!(f, "Encoding error: {}", err),
186        }
187    }
188}
189
190impl std::error::Error for PayloadConversionError {
191    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
192        match self {
193            PayloadConversionError::WrongEncoding => None,
194            PayloadConversionError::EncodingError(err) => Some(err.as_ref()),
195        }
196    }
197}
198
199/// Converts between Rust errors and Temporal [`Failure`] protobufs.
200pub trait FailureConverter {
201    /// Convert an error into a Temporal failure protobuf.
202    fn to_failure(
203        &self,
204        error: Box<dyn std::error::Error>,
205        payload_converter: &PayloadConverter,
206        context: &SerializationContextData,
207    ) -> Result<Failure, PayloadConversionError>;
208
209    /// Convert a Temporal failure protobuf back into a Rust error.
210    fn to_error(
211        &self,
212        failure: Failure,
213        payload_converter: &PayloadConverter,
214        context: &SerializationContextData,
215    ) -> Result<Box<dyn std::error::Error>, PayloadConversionError>;
216}
217/// Default (currently unimplemented) failure converter.
218pub struct DefaultFailureConverter;
219/// Encodes and decodes payloads, enabling encryption or compression.
220pub trait PayloadCodec {
221    /// Encode payloads before they are sent to the server.
222    fn encode(
223        &self,
224        context: &SerializationContextData,
225        payloads: Vec<Payload>,
226    ) -> BoxFuture<'static, Vec<Payload>>;
227    /// Decode payloads after they are received from the server.
228    fn decode(
229        &self,
230        context: &SerializationContextData,
231        payloads: Vec<Payload>,
232    ) -> BoxFuture<'static, Vec<Payload>>;
233}
234
235impl<T: PayloadCodec> PayloadCodec for Arc<T> {
236    fn encode(
237        &self,
238        context: &SerializationContextData,
239        payloads: Vec<Payload>,
240    ) -> BoxFuture<'static, Vec<Payload>> {
241        (**self).encode(context, payloads)
242    }
243    fn decode(
244        &self,
245        context: &SerializationContextData,
246        payloads: Vec<Payload>,
247    ) -> BoxFuture<'static, Vec<Payload>> {
248        (**self).decode(context, payloads)
249    }
250}
251
252/// A no-op codec that passes payloads through unchanged.
253pub struct DefaultPayloadCodec;
254
255/// Indicates some type can be serialized for use with Temporal.
256///
257/// You don't need to implement this unless you are using a non-serde-compatible custom converter,
258/// in which case you should implement the to/from_payload functions on some wrapper type.
259pub trait TemporalSerializable {
260    /// Return a reference to this value as a serde-serializable trait object.
261    fn as_serde(&self) -> Result<&dyn erased_serde::Serialize, PayloadConversionError> {
262        Err(PayloadConversionError::WrongEncoding)
263    }
264    /// Convert this value into a single [`Payload`].
265    fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
266        Err(PayloadConversionError::WrongEncoding)
267    }
268    /// Convert to multiple payloads. Override this for types representing multiple arguments.
269    fn to_payloads(
270        &self,
271        ctx: &SerializationContext<'_>,
272    ) -> Result<Vec<Payload>, PayloadConversionError> {
273        Ok(vec![self.to_payload(ctx)?])
274    }
275}
276
277/// Indicates some type can be deserialized for use with Temporal.
278///
279/// You don't need to implement this unless you are using a non-serde-compatible custom converter,
280/// in which case you should implement the to/from_payload functions on some wrapper type.
281pub trait TemporalDeserializable: Sized {
282    /// Deserialize from a serde-based payload converter.
283    fn from_serde(
284        _: &dyn ErasedSerdePayloadConverter,
285        _ctx: &SerializationContext<'_>,
286        _: Payload,
287    ) -> Result<Self, PayloadConversionError> {
288        Err(PayloadConversionError::WrongEncoding)
289    }
290    /// Deserialize from a single [`Payload`].
291    fn from_payload(
292        ctx: &SerializationContext<'_>,
293        payload: Payload,
294    ) -> Result<Self, PayloadConversionError> {
295        let _ = (ctx, payload);
296        Err(PayloadConversionError::WrongEncoding)
297    }
298    /// Convert from multiple payloads. Override this for types representing multiple arguments.
299    fn from_payloads(
300        ctx: &SerializationContext<'_>,
301        payloads: Vec<Payload>,
302    ) -> Result<Self, PayloadConversionError> {
303        if payloads.len() != 1 {
304            return Err(PayloadConversionError::WrongEncoding);
305        }
306        Self::from_payload(ctx, payloads.into_iter().next().unwrap())
307    }
308}
309
310/// An unconverted set of payloads, used when the caller wants to defer deserialization.
311#[derive(Clone, Debug, Default)]
312pub struct RawValue {
313    /// The underlying payloads.
314    pub payloads: Vec<Payload>,
315}
316impl RawValue {
317    /// A RawValue representing no meaningful data, containing a single default payload.
318    /// This ensures the value can still be serialized as a single payload.
319    pub fn empty() -> Self {
320        Self {
321            payloads: vec![Payload::default()],
322        }
323    }
324
325    /// Create a new RawValue from a vector of payloads.
326    pub fn new(payloads: Vec<Payload>) -> Self {
327        Self { payloads }
328    }
329
330    /// Create a [`RawValue`] by serializing a value with the given converter.
331    pub fn from_value<T: TemporalSerializable + 'static>(
332        value: &T,
333        converter: &PayloadConverter,
334    ) -> RawValue {
335        RawValue::new(vec![
336            converter
337                .to_payload(
338                    &SerializationContext {
339                        data: &SerializationContextData::None,
340                        converter,
341                    },
342                    value,
343                )
344                .unwrap(),
345        ])
346    }
347
348    /// Deserialize this [`RawValue`] into a typed value using the given converter.
349    pub fn to_value<T: TemporalDeserializable + 'static>(self, converter: &PayloadConverter) -> T {
350        converter
351            .from_payload(
352                &SerializationContext {
353                    data: &SerializationContextData::None,
354                    converter,
355                },
356                self.payloads.into_iter().next().unwrap(),
357            )
358            .unwrap()
359    }
360}
361
362impl TemporalSerializable for RawValue {
363    fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
364        Ok(self.payloads.first().cloned().unwrap_or_default())
365    }
366    fn to_payloads(
367        &self,
368        _: &SerializationContext<'_>,
369    ) -> Result<Vec<Payload>, PayloadConversionError> {
370        Ok(self.payloads.clone())
371    }
372}
373
374impl TemporalDeserializable for RawValue {
375    fn from_payload(
376        _: &SerializationContext<'_>,
377        p: Payload,
378    ) -> Result<Self, PayloadConversionError> {
379        Ok(RawValue { payloads: vec![p] })
380    }
381    fn from_payloads(
382        _: &SerializationContext<'_>,
383        payloads: Vec<Payload>,
384    ) -> Result<Self, PayloadConversionError> {
385        Ok(RawValue { payloads })
386    }
387}
388
389/// Generic interface for converting between typed values and [`Payload`]s.
390pub trait GenericPayloadConverter {
391    /// Serialize a value into a single [`Payload`].
392    fn to_payload<T: TemporalSerializable + 'static>(
393        &self,
394        context: &SerializationContext<'_>,
395        val: &T,
396    ) -> Result<Payload, PayloadConversionError>;
397    /// Deserialize a value from a single [`Payload`].
398    #[allow(clippy::wrong_self_convention)]
399    fn from_payload<T: TemporalDeserializable + 'static>(
400        &self,
401        context: &SerializationContext<'_>,
402        payload: Payload,
403    ) -> Result<T, PayloadConversionError>;
404    /// Serialize a value into multiple [`Payload`]s.
405    fn to_payloads<T: TemporalSerializable + 'static>(
406        &self,
407        context: &SerializationContext<'_>,
408        val: &T,
409    ) -> Result<Vec<Payload>, PayloadConversionError> {
410        Ok(vec![self.to_payload(context, val)?])
411    }
412    /// Deserialize a value from multiple [`Payload`]s.
413    #[allow(clippy::wrong_self_convention)]
414    fn from_payloads<T: TemporalDeserializable + 'static>(
415        &self,
416        context: &SerializationContext<'_>,
417        payloads: Vec<Payload>,
418    ) -> Result<T, PayloadConversionError> {
419        if payloads.len() != 1 {
420            return Err(PayloadConversionError::WrongEncoding);
421        }
422        self.from_payload(context, payloads.into_iter().next().unwrap())
423    }
424}
425
426impl GenericPayloadConverter for PayloadConverter {
427    fn to_payload<T: TemporalSerializable + 'static>(
428        &self,
429        context: &SerializationContext<'_>,
430        val: &T,
431    ) -> Result<Payload, PayloadConversionError> {
432        // If a single payload is explicitly needed for `()`, then produce a null payload
433        if std::any::TypeId::of::<T>() == std::any::TypeId::of::<()>() {
434            return Ok(Payload {
435                metadata: {
436                    let mut hm = HashMap::new();
437                    hm.insert("encoding".to_string(), b"binary/null".to_vec());
438                    hm
439                },
440                data: vec![],
441                external_payloads: vec![],
442            });
443        }
444        let mut payloads = self.to_payloads(context, val)?;
445        if payloads.len() != 1 {
446            return Err(PayloadConversionError::WrongEncoding);
447        }
448        Ok(payloads.pop().unwrap())
449    }
450
451    fn from_payload<T: TemporalDeserializable + 'static>(
452        &self,
453        context: &SerializationContext<'_>,
454        payload: Payload,
455    ) -> Result<T, PayloadConversionError> {
456        self.from_payloads(context, vec![payload])
457    }
458
459    fn to_payloads<T: TemporalSerializable + 'static>(
460        &self,
461        context: &SerializationContext<'_>,
462        val: &T,
463    ) -> Result<Vec<Payload>, PayloadConversionError> {
464        match self {
465            PayloadConverter::Serde(pc) => {
466                // Since Rust SDK uses () to denote no input, we must match other SDKs by producing
467                // no payloads for it.
468                if std::any::TypeId::of::<T>() == std::any::TypeId::of::<()>() {
469                    Ok(Vec::new())
470                } else {
471                    Ok(vec![pc.to_payload(context.data, val.as_serde()?)?])
472                }
473            }
474            PayloadConverter::UseWrappers => T::to_payloads(val, context),
475            PayloadConverter::Composite(composite) => {
476                for converter in &composite.converters {
477                    match converter.to_payloads(context, val) {
478                        Ok(payloads) => return Ok(payloads),
479                        Err(PayloadConversionError::WrongEncoding) => continue,
480                        Err(e) => return Err(e),
481                    }
482                }
483                Err(PayloadConversionError::WrongEncoding)
484            }
485        }
486    }
487
488    fn from_payloads<T: TemporalDeserializable + 'static>(
489        &self,
490        context: &SerializationContext<'_>,
491        payloads: Vec<Payload>,
492    ) -> Result<T, PayloadConversionError> {
493        // Accept empty payloads (no args) and a single binary/null payload (result from a
494        // workflow/update with () return type as ().
495        if std::any::TypeId::of::<T>() == std::any::TypeId::of::<()>()
496            && is_unit_payloads(&payloads)
497        {
498            let boxed: Box<dyn std::any::Any> = Box::new(());
499            return Ok(*boxed.downcast::<T>().unwrap());
500        }
501
502        match self {
503            PayloadConverter::Serde(pc) => {
504                if payloads.len() != 1 {
505                    return Err(PayloadConversionError::WrongEncoding);
506                }
507                T::from_serde(pc.as_ref(), context, payloads.into_iter().next().unwrap())
508            }
509            PayloadConverter::UseWrappers => T::from_payloads(context, payloads),
510            PayloadConverter::Composite(composite) => {
511                for converter in &composite.converters {
512                    match converter.from_payloads(context, payloads.clone()) {
513                        Ok(val) => return Ok(val),
514                        Err(PayloadConversionError::WrongEncoding) => continue,
515                        Err(e) => return Err(e),
516                    }
517                }
518                Err(PayloadConversionError::WrongEncoding)
519            }
520        }
521    }
522}
523
524fn is_unit_payloads(payloads: &[Payload]) -> bool {
525    match payloads {
526        [] => true,
527        [payload] => {
528            payload.data.is_empty()
529                && payload
530                    .metadata
531                    .get("encoding")
532                    .map(|encoding| encoding == b"binary/null")
533                    .unwrap_or(false)
534        }
535        _ => false,
536    }
537}
538
539// TODO [rust-sdk-branch]: Potentially allow opt-out / no-serde compile flags
540impl<T> TemporalSerializable for T
541where
542    T: serde::Serialize,
543{
544    fn as_serde(&self) -> Result<&dyn erased_serde::Serialize, PayloadConversionError> {
545        Ok(self)
546    }
547}
548impl<T> TemporalDeserializable for T
549where
550    T: serde::de::DeserializeOwned,
551{
552    fn from_serde(
553        pc: &dyn ErasedSerdePayloadConverter,
554        context: &SerializationContext<'_>,
555        payload: Payload,
556    ) -> Result<Self, PayloadConversionError>
557    where
558        Self: Sized,
559    {
560        let mut de = pc.from_payload(context.data, payload)?;
561        erased_serde::deserialize(&mut de)
562            .map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))
563    }
564}
565
566struct SerdeJsonPayloadConverter;
567impl ErasedSerdePayloadConverter for SerdeJsonPayloadConverter {
568    fn to_payload(
569        &self,
570        _: &SerializationContextData,
571        value: &dyn erased_serde::Serialize,
572    ) -> Result<Payload, PayloadConversionError> {
573        let as_json = serde_json::to_vec(value)
574            .map_err(|e| PayloadConversionError::EncodingError(e.into()))?;
575        Ok(Payload {
576            metadata: {
577                let mut hm = HashMap::new();
578                hm.insert("encoding".to_string(), b"json/plain".to_vec());
579                hm
580            },
581            data: as_json,
582            external_payloads: vec![],
583        })
584    }
585
586    fn from_payload(
587        &self,
588        _: &SerializationContextData,
589        payload: Payload,
590    ) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError> {
591        let encoding = payload.metadata.get("encoding").map(|v| v.as_slice());
592        if encoding != Some(b"json/plain".as_slice()) {
593            return Err(PayloadConversionError::WrongEncoding);
594        }
595        let json_v: serde_json::Value = serde_json::from_slice(&payload.data)
596            .map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))?;
597        Ok(Box::new(<dyn erased_serde::Deserializer>::erase(json_v)))
598    }
599}
600/// Type-erased serde-based payload converter for use behind `dyn` trait objects.
601pub trait ErasedSerdePayloadConverter: Send + Sync {
602    /// Serialize a type-erased serde value into a [`Payload`].
603    fn to_payload(
604        &self,
605        context: &SerializationContextData,
606        value: &dyn erased_serde::Serialize,
607    ) -> Result<Payload, PayloadConversionError>;
608    /// Deserialize a [`Payload`] into a type-erased serde deserializer.
609    #[allow(clippy::wrong_self_convention)]
610    fn from_payload(
611        &self,
612        context: &SerializationContextData,
613        payload: Payload,
614    ) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError>;
615}
616
617// TODO [rust-sdk-branch]: All prost things should be behind a compile flag
618
619/// Wrapper for protobuf messages that implements [`TemporalSerializable`]/[`TemporalDeserializable`]
620/// using `binary/protobuf` encoding.
621pub struct ProstSerializable<T: prost::Message>(pub T);
622impl<T> TemporalSerializable for ProstSerializable<T>
623where
624    T: prost::Message + Default + 'static,
625{
626    fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
627        let as_proto = prost::Message::encode_to_vec(&self.0);
628        Ok(Payload {
629            metadata: {
630                let mut hm = HashMap::new();
631                hm.insert("encoding".to_string(), b"binary/protobuf".to_vec());
632                hm
633            },
634            data: as_proto,
635            external_payloads: vec![],
636        })
637    }
638}
639impl<T> TemporalDeserializable for ProstSerializable<T>
640where
641    T: prost::Message + Default + 'static,
642{
643    fn from_payload(
644        _: &SerializationContext<'_>,
645        p: Payload,
646    ) -> Result<Self, PayloadConversionError>
647    where
648        Self: Sized,
649    {
650        let encoding = p.metadata.get("encoding").map(|v| v.as_slice());
651        if encoding != Some(b"binary/protobuf".as_slice()) {
652            return Err(PayloadConversionError::WrongEncoding);
653        }
654        T::decode(p.data.as_slice())
655            .map(ProstSerializable)
656            .map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))
657    }
658}
659
660/// A payload converter that delegates to an ordered list of inner converters.
661#[derive(Clone)]
662pub struct CompositePayloadConverter {
663    converters: Vec<PayloadConverter>,
664}
665
666impl Default for DataConverter {
667    fn default() -> Self {
668        Self::new(
669            PayloadConverter::default(),
670            DefaultFailureConverter,
671            DefaultPayloadCodec,
672        )
673    }
674}
675impl FailureConverter for DefaultFailureConverter {
676    fn to_failure(
677        &self,
678        _: Box<dyn std::error::Error>,
679        _: &PayloadConverter,
680        _: &SerializationContextData,
681    ) -> Result<Failure, PayloadConversionError> {
682        todo!()
683    }
684    fn to_error(
685        &self,
686        _: Failure,
687        _: &PayloadConverter,
688        _: &SerializationContextData,
689    ) -> Result<Box<dyn std::error::Error>, PayloadConversionError> {
690        todo!()
691    }
692}
693impl PayloadCodec for DefaultPayloadCodec {
694    fn encode(
695        &self,
696        _: &SerializationContextData,
697        payloads: Vec<Payload>,
698    ) -> BoxFuture<'static, Vec<Payload>> {
699        async move { payloads }.boxed()
700    }
701    fn decode(
702        &self,
703        _: &SerializationContextData,
704        payloads: Vec<Payload>,
705    ) -> BoxFuture<'static, Vec<Payload>> {
706        async move { payloads }.boxed()
707    }
708}
709
710/// Represents multiple arguments for workflows/activities that accept more than one argument.
711/// Use this when interoperating with other language SDKs that allow multiple arguments.
712macro_rules! impl_multi_args {
713    ($name:ident; $count:expr; $($idx:tt: $ty:ident),+) => {
714        #[doc = concat!("Wrapper for ", stringify!($count), " typed arguments, enabling multi-arg serialization.")]
715        #[derive(Clone, Debug, PartialEq, Eq)]
716        pub struct $name<$($ty),+>($(pub $ty),+);
717
718        impl<$($ty),+> TemporalSerializable for $name<$($ty),+>
719        where
720            $($ty: TemporalSerializable + 'static),+
721        {
722            fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
723                Err(PayloadConversionError::WrongEncoding)
724            }
725            fn to_payloads(
726                &self,
727                ctx: &SerializationContext<'_>,
728            ) -> Result<Vec<Payload>, PayloadConversionError> {
729                Ok(vec![$(ctx.converter.to_payload(ctx, &self.$idx)?),+])
730            }
731        }
732
733        #[allow(non_snake_case)]
734        impl<$($ty),+> From<($($ty),+,)> for $name<$($ty),+> {
735            fn from(t: ($($ty),+,)) -> Self {
736                $name($(t.$idx),+)
737            }
738        }
739
740        impl<$($ty),+> TemporalDeserializable for $name<$($ty),+>
741        where
742            $($ty: TemporalDeserializable + 'static),+
743        {
744            fn from_payload(_: &SerializationContext<'_>, _: Payload) -> Result<Self, PayloadConversionError> {
745                Err(PayloadConversionError::WrongEncoding)
746            }
747            fn from_payloads(
748                ctx: &SerializationContext<'_>,
749                payloads: Vec<Payload>,
750            ) -> Result<Self, PayloadConversionError> {
751                if payloads.len() != $count {
752                    return Err(PayloadConversionError::WrongEncoding);
753                }
754                let mut iter = payloads.into_iter();
755                Ok($name(
756                    $(ctx.converter.from_payload::<$ty>(ctx, iter.next().unwrap())?),+
757                ))
758            }
759        }
760    };
761}
762
763impl_multi_args!(MultiArgs2; 2; 0: A, 1: B);
764impl_multi_args!(MultiArgs3; 3; 0: A, 1: B, 2: C);
765impl_multi_args!(MultiArgs4; 4; 0: A, 1: B, 2: C, 3: D);
766impl_multi_args!(MultiArgs5; 5; 0: A, 1: B, 2: C, 3: D, 4: E);
767impl_multi_args!(MultiArgs6; 6; 0: A, 1: B, 2: C, 3: D, 4: E, 5: F);
768
769#[cfg(test)]
770mod tests {
771    use super::*;
772
773    #[test]
774    fn test_empty_payloads_as_unit_type() {
775        let converter = PayloadConverter::default();
776        let ctx = SerializationContext {
777            data: &SerializationContextData::Workflow,
778            converter: &converter,
779        };
780
781        let empty_payloads: Vec<Payload> = vec![];
782        let result: Result<(), _> = converter.from_payloads(&ctx, empty_payloads);
783
784        assert!(result.is_ok(), "Empty payloads should deserialize as ()");
785    }
786
787    #[test]
788    fn test_unit_type_roundtrip_serde() {
789        let converter = PayloadConverter::serde_json();
790        let ctx = SerializationContext {
791            data: &SerializationContextData::Workflow,
792            converter: &converter,
793        };
794
795        let payloads = converter.to_payloads(&ctx, &()).unwrap();
796        assert!(payloads.is_empty());
797
798        let result: () = converter.from_payloads(&ctx, payloads).unwrap();
799        assert_eq!(result, ());
800    }
801
802    #[test]
803    fn test_unit_composite_roundtrip() {
804        let converter = PayloadConverter::default();
805        let ctx = SerializationContext {
806            data: &SerializationContextData::Workflow,
807            converter: &converter,
808        };
809
810        let payloads = converter.to_payloads(&ctx, &()).unwrap();
811        assert!(payloads.is_empty());
812
813        let result: () = converter.from_payloads(&ctx, payloads).unwrap();
814        assert_eq!(result, ());
815    }
816
817    #[test]
818    fn test_unit_to_payload_roundtrip() {
819        let converter = PayloadConverter::default();
820        let ctx = SerializationContext {
821            data: &SerializationContextData::Workflow,
822            converter: &converter,
823        };
824
825        let mut payloads = vec![converter.to_payload(&ctx, &()).unwrap()];
826        assert!(is_unit_payloads(&payloads));
827        let result: () = converter
828            .from_payload(&ctx, payloads.pop().unwrap())
829            .unwrap();
830        assert_eq!(result, ());
831    }
832
833    #[test]
834    fn test_unit_use_wrappers_returns_wrong_encoding() {
835        let converter = PayloadConverter::UseWrappers;
836        let ctx = SerializationContext {
837            data: &SerializationContextData::Workflow,
838            converter: &converter,
839        };
840
841        let result = converter.to_payloads(&ctx, &());
842        assert!(
843            matches!(result, Err(PayloadConversionError::WrongEncoding)),
844            "{result:?}"
845        );
846    }
847
848    #[test]
849    fn multi_args_round_trip() {
850        let converter = PayloadConverter::default();
851        let ctx = SerializationContext {
852            data: &SerializationContextData::Workflow,
853            converter: &converter,
854        };
855
856        let args = MultiArgs2("hello".to_string(), 42i32);
857        let payloads = converter.to_payloads(&ctx, &args).unwrap();
858        assert_eq!(payloads.len(), 2);
859
860        let result: MultiArgs2<String, i32> = converter.from_payloads(&ctx, payloads).unwrap();
861        assert_eq!(result, args);
862    }
863
864    #[test]
865    fn multi_args_from_tuple() {
866        let args: MultiArgs2<String, i32> = ("hello".to_string(), 42i32).into();
867        assert_eq!(args, MultiArgs2("hello".to_string(), 42));
868    }
869}
temporalio_common/data_converters.rs

temporalio_common/
data_converters.rs
