temporalio_common_wasm/
data_converters.rs

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

temporalio_common_wasm/
data_converters.rs
