Repository: slawlor/ractor
Pull request #229 (2024-04-29 18:51:26)

Serde blanket implementation feature 
When working with types in outside crates it quickly becomes a nuisance to implement various wrapper types just to be able to implement BytesConvertable. This becomes especially annoying once one has to interface with an existing codebase that has the original types deeply ingrained.

This PR adds an optional feature called "blanket_serde" which replaces the current default implementations of BytesConvertable with a blanket implementation utilizing serde and pot.

In the process a handful of bugs were found, where tests and BytesConvertable implementations were inconsistent in their Deserialization / Serialization.
For example there were cases of Strings into_bytes() function being used for serialization in combination with <String as BytesConvertable>::from_bytes being used for deserialization, this didn't cause any errors since the default implementation similarly just called .into_bytes().

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/slawlor/ractor/pull/229?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) Report
Attention: Patch coverage is `94.18605%` with `5 lines` in your changes are missing coverage. Please review.
> Project coverage is 81.67%. Comparing base [(`8c310e7`)](https://app.codecov.io/gh/slawlor/ractor/commit/8c310e73d1d6b1ceba69053ac65311acce33165d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) to head [(`25a3eb0`)](https://app.codecov.io/gh/slawlor/ractor/pull/229?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).
> Report is 16 commits behind head on main.

| [Files](https://app.codecov.io/gh/slawlor/ractor/pull/229?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) | Patch % | Lines |
|---|---|---|
| [ractor/src/serialization.rs](https://app.codecov.io/gh/slawlor/ractor/pull/229?src=pr&el=tree&filepath=ractor%2Fsrc%2Fserialization.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor#diff-cmFjdG9yL3NyYy9zZXJpYWxpemF0aW9uLnJz) | 93.75% | [5 Missing :warning: ](https://app.codecov.io/gh/slawlor/ractor/pull/229?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #229      +/-   ##
==========================================
+ Coverage   79.73%   81.67%   +1.94%     
==========================================
  Files          50       50              
  Lines        9790     9578     -212     
==========================================
+ Hits         7806     7823      +17     
+ Misses       1984     1755     -229     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/slawlor/ractor/pull/229?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Sean+Lawlor).

stalkopat: Added blanket_serde to the docker integration tests
Changed the docker integration tests to actually use BytesConvertable
Fixed a bug where the pg_group integration test wrongly returned success when the underlying actor paniced / errored


Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 888fbbc..05899ef 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -23,6 +23,9 @@ jobs:
           - name: Test ractor with the `cluster` feature
             package: ractor
             flags: -F cluster
+          - name: Test ractor with the `blanket_serde` feature
+            package: ractor
+            flags: -F blanket_serde
           - name: Test ractor_cluster
             package: ractor_cluster
             # flags: 
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 9fb378d..32e6f4d 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -19,6 +19,7 @@ jobs:
       matrix:
         toolchain: [stable]
         os: [ubuntu]
+        features: [blanket_serde, cluster]
 
     steps:
       - uses: actions/checkout@main
@@ -26,7 +27,7 @@ jobs:
       - name: Build the docker image
         working-directory: .
         run: |
-          docker compose build
+          FEATURES=${{matrix.features}} docker compose build
 
       - name: Authentication Handshake
         working-directory: .
diff --git a/docker-compose.yml b/docker-compose.yml
index c19e599..9319774 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -5,6 +5,8 @@ services:
     build:
       context: .
       dockerfile: ractor_cluster_integration_tests/Dockerfile
+      args:
+        - FEATURES=${FEATURES}
     image: ractor_cluster_tests:latest
     networks:
       - test-net
@@ -19,6 +21,8 @@ services:
     build:
       context: .
       dockerfile: ractor_cluster_integration_tests/Dockerfile
+      args:
+        - FEATURES=${FEATURES}
     image: ractor_cluster_tests:latest
     networks:
       - test-net
@@ -34,6 +38,8 @@ services:
     build:
       context: .
       dockerfile: ractor_cluster_integration_tests/Dockerfile
+      args:
+        - FEATURES=${FEATURES}
     image: ractor_cluster_tests:latest
     networks:
       - test-net
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 8af4e84..aa5d72f 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -17,6 +17,7 @@ rust-version = "1.64"
 ### Other features
 cluster = []
 tokio_runtime = ["tokio/time", "tokio/rt", "tokio/macros", "tokio/tracing"]
+blanket_serde = ["serde", "pot", "cluster"]
 
 # default = ["async-std"]
 default = ["tokio_runtime", "async-trait"]
@@ -35,6 +36,10 @@ async-trait = { version = "0.1", optional = true }
 tokio = { version = "1", features = ["sync"] }
 tracing = { version = "0.1", features = ["attributes"] }
 
+## Blanket Serde
+serde = { version = "1", optional = true }
+pot =  { version = "3.0", optional = true }
+
 [dev-dependencies]
 backtrace = "0.3"
 criterion = "0.5"
diff --git a/ractor/src/factory/job.rs b/ractor/src/factory/job.rs
index 7d15bb9..989a592 100644
--- a/ractor/src/factory/job.rs
+++ b/ractor/src/factory/job.rs
@@ -83,8 +83,8 @@ impl BytesConvertable for JobOptions {
         } else {
             let ttl_bytes = data.split_off(8);
 
-            let submit_time = <u64 as BytesConvertable>::from_bytes(data);
-            let ttl = <u64 as BytesConvertable>::from_bytes(ttl_bytes);
+            let submit_time = u64::from_be_bytes(data.try_into().unwrap()); //Unwrap should be safe since we checked length earlier
+            let ttl = u64::from_be_bytes(ttl_bytes.try_into().unwrap());
 
             Self {
                 submit_time: std::time::UNIX_EPOCH + Duration::from_nanos(submit_time),
@@ -293,14 +293,14 @@ mod tests {
             match self {
                 Self::A(args) => Ok(crate::message::SerializedMessage::Cast {
                     variant: "A".to_string(),
-                    args: args.into_bytes(),
+                    args: <String as BytesConvertable>::into_bytes(args),
                     metadata: None,
                 }),
                 Self::B(args, _reply) => {
                     let (tx, _rx) = crate::concurrency::oneshot();
                     Ok(crate::message::SerializedMessage::Call {
                         variant: "B".to_string(),
-                        args: args.into_bytes(),
+                        args: <String as BytesConvertable>::into_bytes(args),
                         reply: tx.into(),
                         metadata: None,
                     })
@@ -313,7 +313,7 @@ mod tests {
             match bytes {
                 crate::message::SerializedMessage::Cast { variant, args, .. } => {
                     match variant.as_str() {
-                        "A" => Ok(Self::A(String::from_bytes(args))),
+                        "A" => Ok(Self::A(<String as BytesConvertable>::from_bytes(args))),
                         _ => Err(crate::message::BoxedDowncastErr),
                     }
                 }
@@ -321,7 +321,10 @@ mod tests {
                     match variant.as_str() {
                         "B" => {
                             let (tx, _rx) = crate::concurrency::oneshot();
-                            Ok(Self::B(String::from_bytes(args), tx.into()))
+                            Ok(Self::B(
+                                <String as BytesConvertable>::from_bytes(args),
+                                tx.into(),
+                            ))
                         }
                         _ => Err(crate::message::BoxedDowncastErr),
                     }
diff --git a/ractor/src/serialization.rs b/ractor/src/serialization.rs
index a4b6c35..3545249 100644
--- a/ractor/src/serialization.rs
+++ b/ractor/src/serialization.rs
@@ -23,160 +23,181 @@ pub trait BytesConvertable {
     fn from_bytes(bytes: Vec<u8>) -> Self;
 }
 
-// ==================== Primitive implementations ==================== //
-
-macro_rules! implement_numeric {
-    {$ty: ty} => {
-        impl BytesConvertable for $ty {
-            fn into_bytes(self) -> Vec<u8> {
-                self.to_be_bytes().to_vec()
-            }
-            fn from_bytes(bytes: Vec<u8>) -> Self {
-                let mut data = [0u8; std::mem::size_of::<Self>()];
-                data.copy_from_slice(&bytes[..std::mem::size_of::<Self>()]);
-                Self::from_be_bytes(data)
-            }
+#[cfg(feature = "blanket_serde")]
+/// Contains a blanket implementation for all types that implement serde::Serialize and serde::Deserialize
+mod impls {
+    use crate::BytesConvertable;
+
+    impl<T: serde::Serialize + serde::de::DeserializeOwned> BytesConvertable for T {
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            pot::from_slice(&bytes).unwrap()
         }
-    };
+        fn into_bytes(self) -> Vec<u8> {
+            pot::to_vec(&self).unwrap()
+        }
+    }
 }
 
-implement_numeric! {i8}
-implement_numeric! {i16}
-implement_numeric! {i32}
-implement_numeric! {i64}
-implement_numeric! {i128}
-
-implement_numeric! {u8}
-implement_numeric! {u16}
-implement_numeric! {u32}
-implement_numeric! {u64}
-implement_numeric! {u128}
+#[cfg(not(feature = "blanket_serde"))]
+/// Contains the default implementations for the `BytesConvertable` trait
+mod impls {
+    use crate::BytesConvertable;
 
-implement_numeric! {f32}
-implement_numeric! {f64}
+    // ==================== Primitive implementations ==================== //
 
-impl BytesConvertable for () {
-    fn into_bytes(self) -> Vec<u8> {
-        Vec::new()
+    macro_rules! implement_numeric {
+        {$ty: ty} => {
+            impl BytesConvertable for $ty {
+                fn into_bytes(self) -> Vec<u8> {
+                    self.to_be_bytes().to_vec()
+                }
+                fn from_bytes(bytes: Vec<u8>) -> Self {
+                    let mut data = [0u8; std::mem::size_of::<Self>()];
+                    data.copy_from_slice(&bytes[..std::mem::size_of::<Self>()]);
+                    Self::from_be_bytes(data)
+                }
+            }
+        };
     }
-    fn from_bytes(_: Vec<u8>) -> Self {}
-}
 
-impl BytesConvertable for bool {
-    fn into_bytes(self) -> Vec<u8> {
-        if self {
-            vec![1u8]
-        } else {
-            vec![0u8]
+    implement_numeric! {i8}
+    implement_numeric! {i16}
+    implement_numeric! {i32}
+    implement_numeric! {i64}
+    implement_numeric! {i128}
+
+    implement_numeric! {u8}
+    implement_numeric! {u16}
+    implement_numeric! {u32}
+    implement_numeric! {u64}
+    implement_numeric! {u128}
+
+    implement_numeric! {f32}
+    implement_numeric! {f64}
+
+    impl BytesConvertable for () {
+        fn into_bytes(self) -> Vec<u8> {
+            Vec::new()
         }
+        fn from_bytes(_: Vec<u8>) -> Self {}
     }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        bytes[0] == 1u8
-    }
-}
 
-impl BytesConvertable for char {
-    fn into_bytes(self) -> Vec<u8> {
-        let u = self as u32;
-        u.into_bytes()
-    }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        let u = u32::from_bytes(bytes);
-        Self::from_u32(u).unwrap()
+    impl BytesConvertable for bool {
+        fn into_bytes(self) -> Vec<u8> {
+            if self {
+                vec![1u8]
+            } else {
+                vec![0u8]
+            }
+        }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            bytes[0] == 1u8
+        }
     }
-}
 
-impl BytesConvertable for String {
-    fn into_bytes(self) -> Vec<u8> {
-        self.into_bytes()
+    impl BytesConvertable for char {
+        fn into_bytes(self) -> Vec<u8> {
+            let u = self as u32;
+            u.into_bytes()
+        }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            let u = u32::from_bytes(bytes);
+            Self::from_u32(u).unwrap()
+        }
     }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        String::from_utf8(bytes).unwrap()
+
+    impl BytesConvertable for String {
+        fn into_bytes(self) -> Vec<u8> {
+            self.into_bytes()
+        }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            String::from_utf8(bytes).unwrap()
+        }
     }
-}
 
-// ==================== Vectorized implementations ==================== //
+    // ==================== Vectorized implementations ==================== //
 
-macro_rules! implement_vectorized_numeric {
-    {$ty: ty} => {
-        impl BytesConvertable for Vec<$ty> {
-            fn into_bytes(self) -> Vec<u8> {
-                let mut result = vec![0u8; self.len() * std::mem::size_of::<$ty>()];
-                for (offset, item) in self.into_iter().enumerate() {
-                    result[offset * std::mem::size_of::<$ty>() .. offset * std::mem::size_of::<$ty>() + std::mem::size_of::<$ty>()].copy_from_slice(&item.to_be_bytes());
-                }
-                result
-            }
-            fn from_bytes(bytes: Vec<u8>) -> Self {
-                let num_el = bytes.len() / std::mem::size_of::<$ty>();
-                let mut result = vec![<$ty>::MIN; num_el];
-
-                let mut data = [0u8; std::mem::size_of::<$ty>()];
-                for offset in 0..num_el {
-                    data.copy_from_slice(&bytes[offset * std::mem::size_of::<$ty>() .. offset * std::mem::size_of::<$ty>() + std::mem::size_of::<$ty>()]);
-                    result[offset] = <$ty>::from_be_bytes(data);
+    macro_rules! implement_vectorized_numeric {
+        {$ty: ty} => {
+            impl BytesConvertable for Vec<$ty> {
+                fn into_bytes(self) -> Vec<u8> {
+                    let mut result = vec![0u8; self.len() * std::mem::size_of::<$ty>()];
+                    for (offset, item) in self.into_iter().enumerate() {
+                        result[offset * std::mem::size_of::<$ty>() .. offset * std::mem::size_of::<$ty>() + std::mem::size_of::<$ty>()].copy_from_slice(&item.to_be_bytes());
+                    }
+                    result
                 }
+                fn from_bytes(bytes: Vec<u8>) -> Self {
+                    let num_el = bytes.len() / std::mem::size_of::<$ty>();
+                    let mut result = vec![<$ty>::MIN; num_el];
 
-                result
-            }
-        }
-    };
-}
+                    let mut data = [0u8; std::mem::size_of::<$ty>()];
+                    for offset in 0..num_el {
+                        data.copy_from_slice(&bytes[offset * std::mem::size_of::<$ty>() .. offset * std::mem::size_of::<$ty>() + std::mem::size_of::<$ty>()]);
+                        result[offset] = <$ty>::from_be_bytes(data);
+                    }
 
-implement_vectorized_numeric! {i8}
-implement_vectorized_numeric! {i16}
-implement_vectorized_numeric! {i32}
-implement_vectorized_numeric! {i64}
-implement_vectorized_numeric! {i128}
-
-// We explicitly skip u8, as it has a more
-// optimized definition
-impl BytesConvertable for Vec<u8> {
-    fn into_bytes(self) -> Vec<u8> {
-        self
-    }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        bytes
+                    result
+                }
+            }
+        };
     }
-}
-implement_vectorized_numeric! {u16}
-implement_vectorized_numeric! {u32}
-implement_vectorized_numeric! {u64}
-implement_vectorized_numeric! {u128}
-
-implement_vectorized_numeric! {f32}
-implement_vectorized_numeric! {f64}
-
-impl BytesConvertable for Vec<bool> {
-    fn into_bytes(self) -> Vec<u8> {
-        let mut result = vec![0u8; self.len()];
-        for (ptr, item) in self.into_iter().enumerate() {
-            let byte = if item { [1u8] } else { [0u8] };
-            result[ptr..ptr + 1].copy_from_slice(&byte);
+
+    implement_vectorized_numeric! {i8}
+    implement_vectorized_numeric! {i16}
+    implement_vectorized_numeric! {i32}
+    implement_vectorized_numeric! {i64}
+    implement_vectorized_numeric! {i128}
+
+    // We explicitly skip u8, as it has a more
+    // optimized definition
+    impl BytesConvertable for Vec<u8> {
+        fn into_bytes(self) -> Vec<u8> {
+            self
+        }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            bytes
         }
-        result
     }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        let num_el = bytes.len();
-        let mut result = vec![false; num_el];
-        for (ptr, byte) in bytes.into_iter().enumerate() {
-            result[ptr] = byte == 1u8;
+    implement_vectorized_numeric! {u16}
+    implement_vectorized_numeric! {u32}
+    implement_vectorized_numeric! {u64}
+    implement_vectorized_numeric! {u128}
+
+    implement_vectorized_numeric! {f32}
+    implement_vectorized_numeric! {f64}
+
+    impl BytesConvertable for Vec<bool> {
+        fn into_bytes(self) -> Vec<u8> {
+            let mut result = vec![0u8; self.len()];
+            for (ptr, item) in self.into_iter().enumerate() {
+                let byte = if item { [1u8] } else { [0u8] };
+                result[ptr..ptr + 1].copy_from_slice(&byte);
+            }
+            result
         }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            let num_el = bytes.len();
+            let mut result = vec![false; num_el];
+            for (ptr, byte) in bytes.into_iter().enumerate() {
+                result[ptr] = byte == 1u8;
+            }
 
-        result
+            result
+        }
     }
-}
 
-impl BytesConvertable for Vec<char> {
-    fn into_bytes(self) -> Vec<u8> {
-        let data = self.into_iter().map(|c| c as u32).collect::<Vec<_>>();
-        data.into_bytes()
-    }
-    fn from_bytes(bytes: Vec<u8>) -> Self {
-        let u32s = <Vec<u32>>::from_bytes(bytes);
-        u32s.into_iter()
-            .map(|u| char::from_u32(u).unwrap())
-            .collect::<Vec<_>>()
+    impl BytesConvertable for Vec<char> {
+        fn into_bytes(self) -> Vec<u8> {
+            let data = self.into_iter().map(|c| c as u32).collect::<Vec<_>>();
+            data.into_bytes()
+        }
+        fn from_bytes(bytes: Vec<u8>) -> Self {
+            let u32s = <Vec<u32>>::from_bytes(bytes);
+            u32s.into_iter()
+                .map(|u| char::from_u32(u).unwrap())
+                .collect::<Vec<_>>()
+        }
     }
 }
 
@@ -253,7 +274,7 @@ mod tests {
     #[allow(non_snake_case)]
     fn test_bytes_conversion_String() {
         let test_data: String = random_string();
-        let bytes = test_data.clone().into_bytes();
+        let bytes = <String as BytesConvertable>::into_bytes(test_data.clone());
         let back = <String as BytesConvertable>::from_bytes(bytes);
         assert_eq!(test_data, back);
     }
diff --git a/ractor_cluster_derive/src/lib.rs b/ractor_cluster_derive/src/lib.rs
index bf5207e..3fac028 100644
--- a/ractor_cluster_derive/src/lib.rs
+++ b/ractor_cluster_derive/src/lib.rs
@@ -155,7 +155,7 @@ fn impl_variant_serialize(variant: &Variant) -> impl ToTokens {
                     .unnamed
                     .iter()
                     .enumerate()
-                    .map(|(i, _)| format_ident!("field{}", i))
+                    .map(|(i, arg)| (format_ident!("field{}", i), &arg.ty))
                     .collect::<Vec<_>>();
                 // the last field is the port
                 let _ = fields.pop();
@@ -184,9 +184,10 @@ fn impl_variant_serialize(variant: &Variant) -> impl ToTokens {
                         }
                     }
                 } else {
-                    let packed = fields.iter().map(pack_args);
+                    let field_names = fields.iter().map(|(a, _)| a);
+                    let packed = fields.iter().map(|(field, arg)| pack_args(field, arg));
                     quote! {
-                        Self::#name(#(#fields),*, #port) => {
+                        Self::#name(#(#field_names),*, #port) => {
                             let mut data = vec![];
                             #(#packed;)*
                             let target_port = #target_port;
@@ -222,12 +223,13 @@ fn impl_variant_serialize(variant: &Variant) -> impl ToTokens {
                     .unnamed
                     .iter()
                     .enumerate()
-                    .map(|(i, _)| format_ident!("field{}", i))
+                    .map(|(i, arg)| (format_ident!("field{}", i), &arg.ty))
                     .collect::<Vec<_>>();
 
-                let packed = fields.iter().map(pack_args);
+                let field_names = fields.iter().map(|(a, _)| a);
+                let packed = fields.iter().map(|(field, arg)| pack_args(field, arg));
                 quote! {
-                    Self::#name(#(#fields),*) => {
+                    Self::#name(#(#field_names),*) => {
                         let mut data = vec![];
 
                         #(#packed ;)*
@@ -328,10 +330,10 @@ fn impl_call_variant_deserialize(variant: &Variant) -> impl ToTokens {
     }
 }
 
-fn pack_args(field: &Ident) -> impl ToTokens {
+fn pack_args(field: &Ident, target_type: &syn::Type) -> impl ToTokens {
     quote! {
         {
-            let arg_data = #field.into_bytes();
+            let arg_data = <#target_type as ractor::BytesConvertable>::into_bytes(#field);
             let arg_len = (arg_data.len() as u64).to_be_bytes();
             data.extend(arg_len);
             data.extend(arg_data);
@@ -359,6 +361,7 @@ fn convert_deserialize_port(
     the_port: &Ident,
     port_type: &AngleBracketedGenericArguments,
 ) -> impl ToTokens {
+    let generic_args = &port_type.args;
     // TODO: catch unwind for the conversion? returning Err(BoxedDowncastErr)
     quote! {
         {
@@ -367,11 +370,11 @@ fn convert_deserialize_port(
             ractor::concurrency::spawn(async move {
                 if let Some(timeout) = o_timeout {
                     if let Ok(Ok(result)) = ractor::concurrency::timeout(timeout, rx).await {
-                        let _ = #the_port.send(result.into_bytes());
+                        let _ = #the_port.send(<#generic_args as BytesConvertable>::into_bytes(result));
                     }
                 } else {
                     if let Ok(result) = rx.await {
-                        let _ = #the_port.send(result.into_bytes());
+                        let _ = #the_port.send(<#generic_args as BytesConvertable>::into_bytes(result));
                     }
                 }
             });


Test patch:
diff --git a/ractor_cluster_integration_tests/Cargo.toml b/ractor_cluster_integration_tests/Cargo.toml
index 00a3651..ab927c7 100644
--- a/ractor_cluster_integration_tests/Cargo.toml
+++ b/ractor_cluster_integration_tests/Cargo.toml
@@ -10,6 +10,10 @@ repository = "https://github.com/slawlor/ractor"
 readme = "../README.md"
 publish = false
 
+[features]
+blanket_serde = ["ractor/blanket_serde"]
+cluster = []
+
 [dependencies]
 anyhow = "1"
 async-trait = "0.1"
diff --git a/ractor_cluster_integration_tests/Dockerfile b/ractor_cluster_integration_tests/Dockerfile
index e69672e..b433f56 100644
--- a/ractor_cluster_integration_tests/Dockerfile
+++ b/ractor_cluster_integration_tests/Dockerfile
@@ -7,7 +7,8 @@ FROM rust:latest
 WORKDIR /usr/src/app
 RUN apt-get update && apt-get install -y cmake build-essential
 COPY . .
-RUN cargo install --path ractor_cluster_integration_tests/
+ARG FEATURES
+RUN cargo install --path ractor_cluster_integration_tests/ -F $FEATURES
 RUN mv /usr/local/cargo/bin/ractor_cluster_integration_tests /usr/local/bin/ractor_cluster_integration_tests
 RUN mv /usr/src/app/ractor_cluster_integration_tests/test-ca/ /usr/local/bin/test-ca
 WORKDIR /usr/local/bin
diff --git a/ractor_cluster_integration_tests/src/ractor_forward_port_tests.rs b/ractor_cluster_integration_tests/src/ractor_forward_port_tests.rs
index e77cf2b..aa7c6de 100644
--- a/ractor_cluster_integration_tests/src/ractor_forward_port_tests.rs
+++ b/ractor_cluster_integration_tests/src/ractor_forward_port_tests.rs
@@ -34,7 +34,9 @@ async fn no_timeout_rpc() {
     match no_timeout_serialized {
         SerializedMessage::Call { args, reply, .. } => {
             let str = get_len_encoded_string(args);
-            let _ = reply.send(str.into_bytes());
+            let _ = reply.send(<String as ractor_cluster::BytesConvertable>::into_bytes(
+                str,
+            ));
         }
         _ => panic!("Invalid"),
     }
@@ -53,7 +55,9 @@ async fn with_timeout_rpc() {
     match with_timeout_serialized {
         SerializedMessage::Call { args, reply, .. } => {
             let str = get_len_encoded_string(args);
-            let _ = reply.send(str.into());
+            let _ = reply.send(<String as ractor_cluster::BytesConvertable>::into_bytes(
+                str,
+            ));
         }
         _ => panic!("Invalid"),
     }
diff --git a/ractor_cluster_integration_tests/src/tests/pg_groups.rs b/ractor_cluster_integration_tests/src/tests/pg_groups.rs
index 7f90cd4..c77abba 100644
--- a/ractor_cluster_integration_tests/src/tests/pg_groups.rs
+++ b/ractor_cluster_integration_tests/src/tests/pg_groups.rs
@@ -20,27 +20,27 @@ use ractor::concurrency::{sleep, Duration, Instant};
 use ractor::{Actor, ActorProcessingErr, ActorRef, RpcReplyPort};
 use ractor_cluster::RactorClusterMessage;
 
-const NUM_PING_PONGS: u16 = 10;
-const PING_PONG_ALLOTED_MS: u128 = 1500;
+const NUM_HELLOS: u16 = 10;
+const HELLO_ALLOTED_MS: u128 = 1500;
 
-struct PingPongActor;
+struct HelloActor;
 
 #[derive(RactorClusterMessage)]
-enum PingPongActorMessage {
-    Ping,
+enum HelloActorMessage {
+    Hey(String),
     #[rpc]
     IsDone(RpcReplyPort<bool>),
 }
 
-struct PingPongActorState {
+struct HelloActorState {
     count: u16,
     done: bool,
 }
 
 #[ractor::async_trait]
-impl Actor for PingPongActor {
-    type Msg = PingPongActorMessage;
-    type State = PingPongActorState;
+impl Actor for HelloActor {
+    type Msg = HelloActorMessage;
+    type State = HelloActorState;
     type Arguments = ();
 
     async fn pre_start(
@@ -49,7 +49,7 @@ impl Actor for PingPongActor {
         _: (),
     ) -> Result<Self::State, ActorProcessingErr> {
         ractor::pg::join("test".to_string(), vec![myself.get_cell()]);
-        Ok(PingPongActorState {
+        Ok(HelloActorState {
             count: 0,
             done: false,
         })
@@ -68,17 +68,22 @@ impl Actor for PingPongActor {
             .map(ActorRef::<Self::Msg>::from)
             .collect::<Vec<_>>();
         match message {
-            Self::Msg::Ping => {
+            Self::Msg::Hey(message) => {
+                assert!(
+                    message.starts_with("Hey there"),
+                    "Invalid hey message received"
+                );
                 tracing::info!(
-                    "Received a ping, replying in kind to {} remote actors",
+                    "Received a hey {}, replying in kind to {} remote actors",
+                    message,
                     remote_actors.len()
                 );
                 for act in remote_actors {
-                    act.cast(PingPongActorMessage::Ping)?;
+                    act.cast(HelloActorMessage::Hey(message.clone()))?;
                 }
                 state.count += 1;
 
-                if state.count > NUM_PING_PONGS {
+                if state.count > NUM_HELLOS {
                     state.done = true;
                 }
             }
@@ -119,7 +124,7 @@ pub(crate) async fn test(config: PgGroupsConfig) -> i32 {
         .await
         .expect("Failed to start NodeServer A");
 
-    let (test_actor, test_handle) = Actor::spawn(None, PingPongActor, ())
+    let (test_actor, test_handle) = Actor::spawn(None, HelloActor, ())
         .await
         .expect("Ping pong actor failed to start up!");
 
@@ -173,21 +178,21 @@ pub(crate) async fn test(config: PgGroupsConfig) -> i32 {
 
     if err_code == 0 {
         // we're authenticated. Startup our PG group testing
-        let _ = ractor::cast!(test_actor, PingPongActorMessage::Ping);
+        let _ = ractor::cast!(test_actor, HelloActorMessage::Hey("Hey there".to_string()));
         let tic = Instant::now();
 
-        let mut rpc_result = ractor::call_t!(test_actor, PingPongActorMessage::IsDone, 500);
-        while rpc_result.is_ok() {
+        let mut rpc_result = ractor::call_t!(test_actor, HelloActorMessage::IsDone, 500);
+        loop {
             let duration: Duration = Instant::now() - tic;
-            if duration.as_millis() > PING_PONG_ALLOTED_MS {
-                tracing::error!("Ping pong actor didn't complete in allotted time");
+            if duration.as_millis() > HELLO_ALLOTED_MS {
+                tracing::error!("Hello actor didn't complete in allotted time");
                 return -1;
             }
 
             match rpc_result {
                 Ok(true) => {
                     // test completed
-                    tracing::info!("Ping pong actor is completed");
+                    tracing::info!("Hello actor is completed");
                     break;
                 }
                 Ok(_) => {
@@ -200,7 +205,7 @@ pub(crate) async fn test(config: PgGroupsConfig) -> i32 {
                     return -4;
                 }
             }
-            rpc_result = ractor::call_t!(test_actor, PingPongActorMessage::IsDone, 500);
+            rpc_result = ractor::call_t!(test_actor, HelloActorMessage::IsDone, 500);
         }
     } else {
         tracing::warn!("Failed to authenticate, failing test");