mod failure_converter;
mod well_known;
pub use failure_converter::{
ActivityExecutionDecodeHint, CancelExternalWorkflowDecodeHint,
ChildWorkflowExecutionDecodeHint, ChildWorkflowStartDecodeHint, CommonAttributes,
DefaultFailureConverter, FailureConverter, FailureDecodeHint, NoopDecodeHint,
WorkflowSignalDecodeHint,
};
use well_known::{BINARY_NULL_ENCODING_VAL, WellKnownType, binary_null_payload};
use crate::protos::{ENCODING_PAYLOAD_KEY, JSON_ENCODING_VAL, temporal::api::common::v1::Payload};
use futures::{FutureExt, future::BoxFuture};
use std::{collections::HashMap, sync::Arc};
const PROTOBUF_ENCODING_VAL: &str = "binary/protobuf";
#[derive(Clone)]
pub struct DataConverter {
payload_converter: PayloadConverter,
#[allow(dead_code)] failure_converter: Arc<dyn FailureConverter + Send + Sync>,
codec: Arc<dyn PayloadCodec + Send + Sync>,
}
impl std::fmt::Debug for DataConverter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DataConverter")
.field("payload_converter", &self.payload_converter)
.finish_non_exhaustive()
}
}
impl DataConverter {
pub fn new(
payload_converter: PayloadConverter,
failure_converter: impl FailureConverter + Send + Sync + 'static,
codec: impl PayloadCodec + Send + Sync + 'static,
) -> Self {
Self {
payload_converter,
failure_converter: Arc::new(failure_converter),
codec: Arc::new(codec),
}
}
pub async fn to_payload<T: TemporalSerializable + 'static>(
&self,
data: &SerializationContextData,
val: &T,
) -> Result<Payload, PayloadConversionError> {
let context = SerializationContext::new(data, &self.payload_converter);
let payload = self.payload_converter.to_payload(&context, val)?;
let encoded = self.codec.encode(data, vec![payload]).await?;
encoded
.into_iter()
.next()
.ok_or(PayloadConversionError::WrongEncoding)
}
pub async fn from_payload<T: TemporalDeserializable + 'static>(
&self,
data: &SerializationContextData,
payload: Payload,
) -> Result<T, PayloadConversionError> {
let context = SerializationContext::new(data, &self.payload_converter);
let decoded = self.codec.decode(data, vec![payload]).await?;
let payload = decoded
.into_iter()
.next()
.ok_or(PayloadConversionError::WrongEncoding)?;
self.payload_converter.from_payload(&context, payload)
}
pub async fn to_payloads<T: TemporalSerializable + 'static>(
&self,
data: &SerializationContextData,
val: &T,
) -> Result<Vec<Payload>, PayloadConversionError> {
let context = SerializationContext::new(data, &self.payload_converter);
let payloads = self.payload_converter.to_payloads(&context, val)?;
self.codec.encode(data, payloads).await
}
pub async fn from_payloads<T: TemporalDeserializable + 'static>(
&self,
data: &SerializationContextData,
payloads: Vec<Payload>,
) -> Result<T, PayloadConversionError> {
let context = SerializationContext::new(data, &self.payload_converter);
let decoded = self.codec.decode(data, payloads).await?;
self.payload_converter.from_payloads(&context, decoded)
}
pub fn payload_converter(&self) -> &PayloadConverter {
&self.payload_converter
}
pub fn failure_converter(&self) -> &(dyn FailureConverter + Send + Sync) {
self.failure_converter.as_ref()
}
pub fn to_error<H: FailureDecodeHint>(
&self,
context: &SerializationContextData,
failure: crate::protos::temporal::api::failure::v1::Failure,
hint: H,
) -> Result<H::Output, PayloadConversionError> {
let normalized =
self.failure_converter
.to_error(failure, &self.payload_converter, context)?;
Ok(hint.adapt(normalized))
}
pub fn to_failure(
&self,
context: &SerializationContextData,
error: crate::error::OutgoingError,
) -> crate::protos::temporal::api::failure::v1::Failure {
self.failure_converter
.to_failure(error, &self.payload_converter, context)
}
pub fn codec(&self) -> &(dyn PayloadCodec + Send + Sync) {
self.codec.as_ref()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct WorkflowSerializationContext {}
#[allow(clippy::new_without_default)]
impl WorkflowSerializationContext {
pub fn new() -> Self {
Self {}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct ActivitySerializationContext {}
#[allow(clippy::new_without_default)]
impl ActivitySerializationContext {
pub fn new() -> Self {
Self {}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct NexusSerializationContext {}
#[allow(clippy::new_without_default)]
impl NexusSerializationContext {
pub fn new() -> Self {
Self {}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum SerializationContextData {
Workflow(WorkflowSerializationContext),
Activity(ActivitySerializationContext),
Nexus(NexusSerializationContext),
None,
}
#[derive(Clone, Copy)]
#[non_exhaustive]
pub struct SerializationContext<'a> {
pub data: &'a SerializationContextData,
pub converter: &'a PayloadConverter,
}
impl<'a> SerializationContext<'a> {
pub fn new(data: &'a SerializationContextData, converter: &'a PayloadConverter) -> Self {
Self { data, converter }
}
}
#[derive(Clone)]
#[non_exhaustive]
pub enum PayloadConverter {
Serde(Arc<dyn ErasedSerdePayloadConverter>),
UseWrappers,
Composite(Arc<CompositePayloadConverter>),
}
impl std::fmt::Debug for PayloadConverter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PayloadConverter::Serde(_) => write!(f, "PayloadConverter::Serde(...)"),
PayloadConverter::UseWrappers => write!(f, "PayloadConverter::UseWrappers"),
PayloadConverter::Composite(_) => write!(f, "PayloadConverter::Composite(...)"),
}
}
}
impl PayloadConverter {
pub fn serde_json() -> Self {
Self::Serde(Arc::new(SerdeJsonPayloadConverter))
}
}
impl Default for PayloadConverter {
fn default() -> Self {
Self::Composite(Arc::new(CompositePayloadConverter {
converters: vec![Self::UseWrappers, Self::serde_json()],
}))
}
}
#[derive(Debug)]
#[non_exhaustive]
pub enum PayloadConversionError {
WrongEncoding,
EncodingError(Box<dyn std::error::Error + Send + Sync>),
}
impl std::fmt::Display for PayloadConversionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PayloadConversionError::WrongEncoding => write!(f, "Wrong encoding"),
PayloadConversionError::EncodingError(err) => write!(f, "Encoding error: {}", err),
}
}
}
impl std::error::Error for PayloadConversionError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
PayloadConversionError::WrongEncoding => None,
PayloadConversionError::EncodingError(err) => Some(err.as_ref()),
}
}
}
pub trait PayloadCodec {
fn encode(
&self,
context: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>>;
fn decode(
&self,
context: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>>;
}
impl<T: PayloadCodec> PayloadCodec for Arc<T> {
fn encode(
&self,
context: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>> {
(**self).encode(context, payloads)
}
fn decode(
&self,
context: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>> {
(**self).decode(context, payloads)
}
}
pub struct DefaultPayloadCodec;
pub trait TemporalSerializable {
fn as_serde(&self) -> Result<&dyn erased_serde::Serialize, PayloadConversionError> {
Err(PayloadConversionError::WrongEncoding)
}
fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
Err(PayloadConversionError::WrongEncoding)
}
fn to_payloads(
&self,
ctx: &SerializationContext<'_>,
) -> Result<Vec<Payload>, PayloadConversionError> {
Ok(vec![self.to_payload(ctx)?])
}
}
pub trait TemporalDeserializable: Sized {
fn from_serde(
_: &dyn ErasedSerdePayloadConverter,
_ctx: &SerializationContext<'_>,
_: Payload,
) -> Result<Self, PayloadConversionError> {
Err(PayloadConversionError::WrongEncoding)
}
fn from_payload(
ctx: &SerializationContext<'_>,
payload: Payload,
) -> Result<Self, PayloadConversionError> {
let _ = (ctx, payload);
Err(PayloadConversionError::WrongEncoding)
}
fn from_payloads(
ctx: &SerializationContext<'_>,
payloads: Vec<Payload>,
) -> Result<Self, PayloadConversionError> {
if payloads.len() != 1 {
return Err(PayloadConversionError::WrongEncoding);
}
Self::from_payload(ctx, payloads.into_iter().next().unwrap())
}
}
#[derive(Clone, Debug)]
pub struct DecodablePayloads {
payloads: Vec<Payload>,
payload_converter: PayloadConverter,
context: SerializationContextData,
}
impl DecodablePayloads {
pub fn new(
payloads: Vec<Payload>,
payload_converter: PayloadConverter,
context: SerializationContextData,
) -> Self {
Self {
payloads,
payload_converter,
context,
}
}
pub fn deserialize<T: TemporalDeserializable + 'static>(
&self,
) -> Result<T, PayloadConversionError> {
self.payload_converter.from_payloads(
&SerializationContext::new(&self.context, &self.payload_converter),
self.payloads.clone(),
)
}
pub fn raw(&self) -> &[Payload] {
&self.payloads
}
pub fn into_raw(self) -> RawValue {
RawValue::new(self.payloads)
}
}
#[derive(Clone, Debug, Default)]
pub struct RawValue {
pub payloads: Vec<Payload>,
}
impl RawValue {
pub fn empty() -> Self {
Self {
payloads: vec![Payload::default()],
}
}
pub fn new(payloads: Vec<Payload>) -> Self {
Self { payloads }
}
pub fn from_value<T: TemporalSerializable + 'static>(
value: &T,
converter: &PayloadConverter,
) -> RawValue {
RawValue::new(vec![
converter
.to_payload(
&SerializationContext::new(&SerializationContextData::None, converter),
value,
)
.unwrap(),
])
}
pub fn to_value<T: TemporalDeserializable + 'static>(self, converter: &PayloadConverter) -> T {
converter
.from_payload(
&SerializationContext::new(&SerializationContextData::None, converter),
self.payloads.into_iter().next().unwrap(),
)
.unwrap()
}
}
impl TemporalSerializable for RawValue {
fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
Ok(self.payloads.first().cloned().unwrap_or_default())
}
fn to_payloads(
&self,
_: &SerializationContext<'_>,
) -> Result<Vec<Payload>, PayloadConversionError> {
Ok(self.payloads.clone())
}
}
impl TemporalDeserializable for RawValue {
fn from_payload(
_: &SerializationContext<'_>,
p: Payload,
) -> Result<Self, PayloadConversionError> {
Ok(RawValue { payloads: vec![p] })
}
fn from_payloads(
_: &SerializationContext<'_>,
payloads: Vec<Payload>,
) -> Result<Self, PayloadConversionError> {
Ok(RawValue { payloads })
}
}
pub trait GenericPayloadConverter {
fn to_payload<T: TemporalSerializable + 'static>(
&self,
context: &SerializationContext<'_>,
val: &T,
) -> Result<Payload, PayloadConversionError>;
#[allow(clippy::wrong_self_convention)]
fn from_payload<T: TemporalDeserializable + 'static>(
&self,
context: &SerializationContext<'_>,
payload: Payload,
) -> Result<T, PayloadConversionError>;
fn to_payloads<T: TemporalSerializable + 'static>(
&self,
context: &SerializationContext<'_>,
val: &T,
) -> Result<Vec<Payload>, PayloadConversionError> {
Ok(vec![self.to_payload(context, val)?])
}
#[allow(clippy::wrong_self_convention)]
fn from_payloads<T: TemporalDeserializable + 'static>(
&self,
context: &SerializationContext<'_>,
payloads: Vec<Payload>,
) -> Result<T, PayloadConversionError> {
if payloads.len() != 1 {
return Err(PayloadConversionError::WrongEncoding);
}
self.from_payload(context, payloads.into_iter().next().unwrap())
}
}
impl GenericPayloadConverter for PayloadConverter {
fn to_payload<T: TemporalSerializable + 'static>(
&self,
context: &SerializationContext<'_>,
val: &T,
) -> Result<Payload, PayloadConversionError> {
match self {
PayloadConverter::Serde(pc) => {
if let Some(well_known_type) = WellKnownType::of::<T>() {
Ok(well_known_type.to_payload(val))
} else {
pc.to_payload(context.data, val.as_serde()?)
}
}
PayloadConverter::UseWrappers => T::to_payload(val, context),
PayloadConverter::Composite(composite) => {
for converter in &composite.converters {
match converter.to_payload(context, val) {
Ok(payload) => return Ok(payload),
Err(PayloadConversionError::WrongEncoding) => continue,
Err(e) => return Err(e),
}
}
Err(PayloadConversionError::WrongEncoding)
}
}
}
fn from_payload<T: TemporalDeserializable + 'static>(
&self,
context: &SerializationContext<'_>,
mut payload: Payload,
) -> Result<T, PayloadConversionError> {
match self {
PayloadConverter::Serde(pc) => {
if let Some(well_known_type) = WellKnownType::of::<T>() {
payload = match well_known_type.try_from_payload(payload) {
Ok(value) => return Ok(value),
Err(payload) => payload,
};
}
T::from_serde(pc.as_ref(), context, payload)
}
PayloadConverter::UseWrappers => T::from_payload(context, payload),
PayloadConverter::Composite(composite) => {
for converter in &composite.converters {
match converter.from_payload(context, payload.clone()) {
Ok(value) => return Ok(value),
Err(PayloadConversionError::WrongEncoding) => continue,
Err(e) => return Err(e),
}
}
Err(PayloadConversionError::WrongEncoding)
}
}
}
fn to_payloads<T: TemporalSerializable + 'static>(
&self,
context: &SerializationContext<'_>,
val: &T,
) -> Result<Vec<Payload>, PayloadConversionError> {
match self {
PayloadConverter::Serde(pc) => {
if let Some(well_known_type) = WellKnownType::of::<T>() {
Ok(well_known_type.to_payloads(val))
} else {
Ok(vec![pc.to_payload(context.data, val.as_serde()?)?])
}
}
PayloadConverter::UseWrappers => T::to_payloads(val, context),
PayloadConverter::Composite(composite) => {
for converter in &composite.converters {
match converter.to_payloads(context, val) {
Ok(payloads) => return Ok(payloads),
Err(PayloadConversionError::WrongEncoding) => continue,
Err(e) => return Err(e),
}
}
Err(PayloadConversionError::WrongEncoding)
}
}
}
fn from_payloads<T: TemporalDeserializable + 'static>(
&self,
context: &SerializationContext<'_>,
mut payloads: Vec<Payload>,
) -> Result<T, PayloadConversionError> {
match self {
PayloadConverter::Serde(pc) => {
if let Some(well_known_type) = WellKnownType::of::<T>() {
payloads = match well_known_type.try_from_payloads(payloads) {
Ok(value) => return Ok(value),
Err(payloads) => payloads,
};
}
if payloads.len() != 1 {
return Err(PayloadConversionError::WrongEncoding);
}
let payload = payloads.into_iter().next().unwrap();
T::from_serde(pc.as_ref(), context, payload)
}
PayloadConverter::UseWrappers => T::from_payloads(context, payloads),
PayloadConverter::Composite(composite) => {
for converter in &composite.converters {
match converter.from_payloads(context, payloads.clone()) {
Ok(val) => return Ok(val),
Err(PayloadConversionError::WrongEncoding) => continue,
Err(e) => return Err(e),
}
}
Err(PayloadConversionError::WrongEncoding)
}
}
}
}
impl<T> TemporalSerializable for T
where
T: serde::Serialize,
{
fn as_serde(&self) -> Result<&dyn erased_serde::Serialize, PayloadConversionError> {
Ok(self)
}
}
impl<T> TemporalDeserializable for T
where
T: serde::de::DeserializeOwned,
{
fn from_serde(
pc: &dyn ErasedSerdePayloadConverter,
context: &SerializationContext<'_>,
payload: Payload,
) -> Result<Self, PayloadConversionError>
where
Self: Sized,
{
let mut de = pc.from_payload(context.data, payload)?;
erased_serde::deserialize(&mut de)
.map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))
}
}
struct SerdeJsonPayloadConverter;
impl ErasedSerdePayloadConverter for SerdeJsonPayloadConverter {
fn to_payload(
&self,
_: &SerializationContextData,
value: &dyn erased_serde::Serialize,
) -> Result<Payload, PayloadConversionError> {
let as_json = serde_json::to_vec(value)
.map_err(|e| PayloadConversionError::EncodingError(e.into()))?;
if as_json.as_slice() == b"null" {
return Ok(binary_null_payload());
}
Ok(Payload {
metadata: {
let mut hm = HashMap::new();
hm.insert(
ENCODING_PAYLOAD_KEY.to_string(),
JSON_ENCODING_VAL.as_bytes().to_vec(),
);
hm
},
data: as_json,
external_payloads: vec![],
})
}
fn from_payload(
&self,
_: &SerializationContextData,
payload: Payload,
) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError> {
let encoding = payload
.metadata
.get(ENCODING_PAYLOAD_KEY)
.map(|v| v.as_slice());
let json_v = if encoding == Some(JSON_ENCODING_VAL.as_bytes()) {
serde_json::from_slice(&payload.data)
.map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))?
} else if encoding == Some(BINARY_NULL_ENCODING_VAL.as_bytes()) {
serde_json::Value::Null
} else {
return Err(PayloadConversionError::WrongEncoding);
};
Ok(Box::new(<dyn erased_serde::Deserializer>::erase(json_v)))
}
}
pub trait ErasedSerdePayloadConverter: Send + Sync {
fn to_payload(
&self,
context: &SerializationContextData,
value: &dyn erased_serde::Serialize,
) -> Result<Payload, PayloadConversionError>;
#[allow(clippy::wrong_self_convention)]
fn from_payload(
&self,
context: &SerializationContextData,
payload: Payload,
) -> Result<Box<dyn erased_serde::Deserializer<'static>>, PayloadConversionError>;
}
pub struct ProstSerializable<T: prost::Message>(pub T);
impl<T> TemporalSerializable for ProstSerializable<T>
where
T: prost::Message + Default + 'static,
{
fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
let as_proto = prost::Message::encode_to_vec(&self.0);
Ok(Payload {
metadata: {
let mut hm = HashMap::new();
hm.insert(
ENCODING_PAYLOAD_KEY.to_string(),
PROTOBUF_ENCODING_VAL.as_bytes().to_vec(),
);
hm
},
data: as_proto,
external_payloads: vec![],
})
}
}
impl<T> TemporalDeserializable for ProstSerializable<T>
where
T: prost::Message + Default + 'static,
{
fn from_payload(
_: &SerializationContext<'_>,
p: Payload,
) -> Result<Self, PayloadConversionError>
where
Self: Sized,
{
let encoding = p.metadata.get(ENCODING_PAYLOAD_KEY).map(|v| v.as_slice());
if encoding != Some(PROTOBUF_ENCODING_VAL.as_bytes()) {
return Err(PayloadConversionError::WrongEncoding);
}
T::decode(p.data.as_slice())
.map(ProstSerializable)
.map_err(|e| PayloadConversionError::EncodingError(Box::new(e)))
}
}
#[derive(Clone)]
pub struct CompositePayloadConverter {
converters: Vec<PayloadConverter>,
}
impl Default for DataConverter {
fn default() -> Self {
Self::new(
PayloadConverter::default(),
DefaultFailureConverter::default(),
DefaultPayloadCodec,
)
}
}
impl PayloadCodec for DefaultPayloadCodec {
fn encode(
&self,
_: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>> {
async move { Ok(payloads) }.boxed()
}
fn decode(
&self,
_: &SerializationContextData,
payloads: Vec<Payload>,
) -> BoxFuture<'static, Result<Vec<Payload>, PayloadConversionError>> {
async move { Ok(payloads) }.boxed()
}
}
macro_rules! impl_multi_args {
($name:ident; $count:expr; $($idx:tt: $ty:ident),+) => {
#[doc = concat!("Wrapper for ", stringify!($count), " typed arguments, enabling multi-arg serialization.")]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct $name<$($ty),+>($(pub $ty),+);
impl<$($ty),+> TemporalSerializable for $name<$($ty),+>
where
$($ty: TemporalSerializable + 'static),+
{
fn to_payload(&self, _: &SerializationContext<'_>) -> Result<Payload, PayloadConversionError> {
Err(PayloadConversionError::WrongEncoding)
}
fn to_payloads(
&self,
ctx: &SerializationContext<'_>,
) -> Result<Vec<Payload>, PayloadConversionError> {
Ok(vec![$(ctx.converter.to_payload(ctx, &self.$idx)?),+])
}
}
#[allow(non_snake_case)]
impl<$($ty),+> From<($($ty),+,)> for $name<$($ty),+> {
fn from(t: ($($ty),+,)) -> Self {
$name($(t.$idx),+)
}
}
impl<$($ty),+> TemporalDeserializable for $name<$($ty),+>
where
$($ty: TemporalDeserializable + 'static),+
{
fn from_payload(_: &SerializationContext<'_>, _: Payload) -> Result<Self, PayloadConversionError> {
Err(PayloadConversionError::WrongEncoding)
}
fn from_payloads(
ctx: &SerializationContext<'_>,
payloads: Vec<Payload>,
) -> Result<Self, PayloadConversionError> {
if payloads.len() != $count {
return Err(PayloadConversionError::WrongEncoding);
}
let mut iter = payloads.into_iter();
Ok($name(
$(ctx.converter.from_payload::<$ty>(ctx, iter.next().unwrap())?),+
))
}
}
};
}
impl_multi_args!(MultiArgs2; 2; 0: A, 1: B);
impl_multi_args!(MultiArgs3; 3; 0: A, 1: B, 2: C);
impl_multi_args!(MultiArgs4; 4; 0: A, 1: B, 2: C, 3: D);
impl_multi_args!(MultiArgs5; 5; 0: A, 1: B, 2: C, 3: D, 4: E);
impl_multi_args!(MultiArgs6; 6; 0: A, 1: B, 2: C, 3: D, 4: E, 5: F);
#[cfg(test)]
mod tests {
use super::*;
use crate::data_converters::well_known::BINARY_PLAIN_ENCODING_VAL;
use rstest::rstest;
#[test]
fn unit_payloads_roundtrip() {
let converter = PayloadConverter::serde_json();
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let payloads = converter.to_payloads(&ctx, &()).unwrap();
assert!(payloads.is_empty());
let result: () = converter.from_payloads(&ctx, payloads).unwrap();
assert_eq!(result, ());
}
#[rstest]
#[case::unit((), BINARY_NULL_ENCODING_VAL, b"")]
#[case::none_string(Option::<String>::None, BINARY_NULL_ENCODING_VAL, b"")]
#[case::some_string(
Some("value".to_string()),
JSON_ENCODING_VAL,
br#""value""#
)]
#[case::bytes(vec![0_u8, 1, 2, 255], BINARY_PLAIN_ENCODING_VAL, &[0, 1, 2, 255])]
#[case::some_bytes(
Some(vec![1_u8, 2, 3]),
BINARY_PLAIN_ENCODING_VAL,
&[1, 2, 3]
)]
#[case::none_bytes(Option::<Vec<u8>>::None, BINARY_NULL_ENCODING_VAL, b"")]
fn value_encodes_as<T>(
#[case] value: T,
#[case] expected_encoding: &str,
#[case] expected_data: &[u8],
) where
T: TemporalSerializable + std::fmt::Debug + 'static,
{
let converter = PayloadConverter::default();
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let payload = converter.to_payload(&ctx, &value).unwrap();
assert_eq!(
payload.metadata.get(ENCODING_PAYLOAD_KEY).unwrap(),
expected_encoding.as_bytes()
);
assert_eq!(payload.data, expected_data);
}
#[rstest]
#[case::unit(BINARY_NULL_ENCODING_VAL, b"", ())]
#[case::none_string(BINARY_NULL_ENCODING_VAL, b"", Option::<String>::None)]
#[case::legacy_none_string(JSON_ENCODING_VAL, b"null", Option::<String>::None)]
#[case::bytes(BINARY_PLAIN_ENCODING_VAL, &[0, 1, 2, 255], vec![0_u8, 1, 2, 255])]
#[case::legacy_bytes(JSON_ENCODING_VAL, b"[3,2,1]", vec![3_u8, 2, 1])]
#[case::some_bytes(
BINARY_PLAIN_ENCODING_VAL,
&[1, 2, 3],
Some(vec![1_u8, 2, 3])
)]
#[case::none_bytes(BINARY_NULL_ENCODING_VAL, b"", Option::<Vec<u8>>::None)]
#[case::legacy_some_bytes(
JSON_ENCODING_VAL,
b"[3,2,1]",
Some(vec![3_u8, 2, 1])
)]
#[case::legacy_none_bytes(JSON_ENCODING_VAL, b"null", Option::<Vec<u8>>::None)]
fn payload_decodes_as<T>(#[case] encoding: &str, #[case] data: &[u8], #[case] expected: T)
where
T: TemporalDeserializable + std::fmt::Debug + PartialEq + 'static,
{
let converter = PayloadConverter::default();
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let actual: T = converter
.from_payload(
&ctx,
Payload {
metadata: HashMap::from([(
ENCODING_PAYLOAD_KEY.to_string(),
encoding.as_bytes().to_vec(),
)]),
data: data.to_vec(),
external_payloads: vec![],
},
)
.unwrap();
assert_eq!(actual, expected);
}
#[test]
fn use_wrappers_returns_wrong_encoding_for_standard_types() {
let converter = PayloadConverter::UseWrappers;
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let result = converter.to_payload(&ctx, &());
assert!(
matches!(result, Err(PayloadConversionError::WrongEncoding)),
"{result:?}"
);
let result = converter.to_payloads(&ctx, &());
assert!(
matches!(result, Err(PayloadConversionError::WrongEncoding)),
"{result:?}"
);
let result = converter.to_payloads(&ctx, &vec![1_u8, 2, 3]);
assert!(
matches!(result, Err(PayloadConversionError::WrongEncoding)),
"{result:?}"
);
let result: Result<(), _> = converter.from_payload(&ctx, binary_null_payload());
assert!(
matches!(result, Err(PayloadConversionError::WrongEncoding)),
"{result:?}"
);
}
#[test]
fn multi_args_round_trip() {
let converter = PayloadConverter::default();
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let args = MultiArgs2("hello".to_string(), 42i32);
let payloads = converter.to_payloads(&ctx, &args).unwrap();
assert_eq!(payloads.len(), 2);
let result: MultiArgs2<String, i32> = converter.from_payloads(&ctx, payloads).unwrap();
assert_eq!(result, args);
}
#[test]
fn empty_payloads_do_not_decode_as_option() {
let converter = PayloadConverter::default();
let context_data = SerializationContextData::Workflow(WorkflowSerializationContext::new());
let ctx = SerializationContext::new(&context_data, &converter);
let result: Result<Option<String>, _> = converter.from_payloads(&ctx, vec![]);
assert!(matches!(result, Err(PayloadConversionError::WrongEncoding)));
}
#[test]
fn multi_args_from_tuple() {
let args: MultiArgs2<String, i32> = ("hello".to_string(), 42i32).into();
assert_eq!(args, MultiArgs2("hello".to_string(), 42));
}
#[rstest]
#[case::string("hello".to_string())]
#[case::some_string(Some("hello".to_string()))]
#[case::none_string(Option::<String>::None)]
#[case::unit(())]
#[case::strings(vec!["hello".to_string(), "world".to_string()])]
#[case::bytes(vec![1_u8, 2, 3])]
#[case::some_bytes(Some(vec![1_u8, 2, 3]))]
#[case::none_bytes(Option::<Vec<u8>>::None)]
fn decodable_payloads_roundtrip<T>(#[case] value: T)
where
T: TemporalSerializable + TemporalDeserializable + std::fmt::Debug + PartialEq + 'static,
{
let converter = PayloadConverter::default();
let payloads = converter
.to_payloads(
&SerializationContext::new(
&SerializationContextData::Workflow(WorkflowSerializationContext::new()),
&converter,
),
&value,
)
.unwrap();
let payloads = DecodablePayloads::new(
payloads,
converter,
SerializationContextData::Workflow(WorkflowSerializationContext::new()),
);
let result: T = payloads.deserialize().unwrap();
assert_eq!(result, value);
}
}