Fix: use json payloads

CHANGELOG.md

@@ -21,10 +21,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   other instances to do the same. Also makes `Client::wait_network_initialized`
   private, as it is included in `Client::new_and_init` now. See [PR 25].
 
+- Use a JSON Payload instead of string in `publish` and `subscribe`. See [PR 34].
+
 [PR 26]: https://github.com/testground/sdk-rust/pull/26
 [PR 25]: https://github.com/testground/sdk-rust/pull/25
 [PR 27]: https://github.com/testground/sdk-rust/pull/27
 [PR 29]: https://github.com/testground/sdk-rust/pull/29
+[PR 34]: https://github.com/testground/sdk-rust/pull/34
 
 ## [0.3.0]
 ### Added

examples/example.rs

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let client = testground::client::Client::new_and_init().await?;
@@ -11,6 +13,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .unwrap()
     ));
 
+    let json = serde_json::json!({"foo": "bar"});
+
+    client.publish("demonstration", Cow::Owned(json)).await?;
+
     client.record_success().await?;
 
     Ok(())

src/background.rs

@@ -23,12 +23,12 @@ const WEBSOCKET_RECEIVER: &str = "Websocket Receiver";
 pub enum Command {
     Publish {
         topic: String,
-        message: String,
+        message: serde_json::Value,
         sender: oneshot::Sender<Result<u64, Error>>,
     },
     Subscribe {
         topic: String,
-        stream: mpsc::Sender<Result<String, Error>>,
+        stream: mpsc::Sender<Result<serde_json::Value, Error>>,
     },
 
     SignalEntry {
@@ -89,7 +89,7 @@ enum PendingRequest {
         sender: oneshot::Sender<Result<(), Error>>,
     },
     Subscribe {
-        stream: mpsc::Sender<Result<String, Error>>,
+        stream: mpsc::Sender<Result<serde_json::Value, Error>>,
     },
 }
 
@@ -233,7 +233,7 @@ impl BackgroundTask {
             } => {
                 let topic = self.contextualize_topic(&topic);
 
-                self.publish(id, topic, PayloadType::String(message), sender)
+                self.publish(id, topic, PayloadType::Json(message), sender)
                     .await
             }
             Command::Subscribe { topic, stream } => {
@@ -406,7 +406,7 @@ impl BackgroundTask {
         &mut self,
         id: u64,
         topic: String,
-        stream: mpsc::Sender<Result<String, Error>>,
+        stream: mpsc::Sender<Result<serde_json::Value, Error>>,
     ) {
         let request = Request {
             id: id.to_string(),

src/client.rs

@@ -87,7 +87,7 @@ impl Client {
     pub async fn publish(
         &self,
         topic: impl Into<Cow<'static, str>>,
-        message: impl Into<Cow<'static, str>>,
+        message: impl Into<Cow<'static, serde_json::Value>>,
     ) -> Result<u64, Error> {
         let (sender, receiver) = oneshot::channel();
 
@@ -106,7 +106,7 @@ impl Client {
     pub async fn subscribe(
         &self,
         topic: impl Into<Cow<'static, str>>,
-    ) -> impl Stream<Item = Result<String, Error>> {
+    ) -> impl Stream<Item = Result<serde_json::Value, Error>> {
         let (stream, out) = mpsc::channel(1);
 
         let cmd = Command::Subscribe {

src/requests.rs

@@ -17,7 +17,7 @@ pub struct Request {
 pub enum PayloadType {
     Event(EventType),
 
-    String(String),
+    Json(serde_json::Value),
 
     Config(NetworkConfiguration),
 }

src/responses.rs

@@ -1,6 +1,24 @@
-use serde::Deserialize;
+use serde::{Deserialize, Deserializer};
 use serde_with::rust::string_empty_as_none;
 
+fn json_empty_string_is_none<'de, D>(data: D) -> Result<Option<serde_json::Value>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value: Option<serde_json::Value> = Deserialize::deserialize(data)?;
+
+    match value {
+        Some(v) => {
+            if v.is_string() && v.as_str().unwrap().is_empty() {
+                Ok(None)
+            } else {
+                Ok(Some(v))
+            }
+        }
+        None => Ok(None),
+    }
+}
+
 #[derive(Deserialize, Debug)]
 pub struct SignalEntry {
     pub seq: u64,
@@ -18,8 +36,8 @@ pub struct RawResponse {
     #[serde(with = "string_empty_as_none")]
     pub error: Option<String>,
 
-    #[serde(with = "string_empty_as_none")]
-    pub subscribe: Option<String>,
+    #[serde(deserialize_with = "json_empty_string_is_none")]
+    pub subscribe: Option<serde_json::Value>,
 
     pub signal_entry: Option<SignalEntry>,
 
@@ -30,7 +48,7 @@ pub struct RawResponse {
 pub enum ResponseType {
     SignalEntry { seq: u64 },
     Publish { seq: u64 },
-    Subscribe(String),
+    Subscribe(serde_json::Value),
     Error(String),
     Barrier,
 }
@@ -58,11 +76,7 @@ impl From<RawResponse> for Response {
                 let error = serde_json::from_str(&error).expect("JSON Deserialization");
                 ResponseType::Error(error)
             }
-            (None, Some(msg), None, None) => {
-                //Hack to remove extra escape characters
-                let msg = serde_json::from_str(&msg).expect("JSON Deserialization");
-                ResponseType::Subscribe(msg)
-            }
+            (None, Some(msg), None, None) => ResponseType::Subscribe(msg),
             (None, None, Some(signal), None) => ResponseType::SignalEntry { seq: signal.seq },
             (None, None, None, Some(publish)) => ResponseType::Publish { seq: publish.seq },
             (error, subscribe, signal_entry, publish) => {
@@ -89,7 +103,8 @@ mod tests {
 
     #[test]
     fn serde_test() {
-        let raw_response = "{\"id\":\"0\",\"error\":\"\",\"subscribe\":\"\",\"publish\":{\"seq\":1},\"signal_entry\":null}";
+        let raw_response =
+            "{\"id\":\"0\",\"error\":\"\",\"subscribe\":\"\",\"publish\":{\"seq\":1},\"signal_entry\":null}";
 
         let response: RawResponse = serde_json::from_str(raw_response).unwrap();
 
@@ -103,4 +118,23 @@ mod tests {
             response
         );
     }
+    #[test]
+    fn serde_test_complex_subscribe() {
+        let raw_response = "{\"id\":\"1\",\"error\":\"\",\"subscribe\":{\"Addrs\":[\"/ip4/16.3.0.3/tcp/45369\"],\"ID\":\"QmbSLMEMackm7vHiUGMB2EFAPbzeJNpeB9yTpzYKoojDWc\"}}";
+
+        let response: RawResponse = serde_json::from_str(raw_response).unwrap();
+
+        let response: Response = response.into();
+
+        assert_eq!(
+            Response {
+                id: "1".to_owned(),
+                response: ResponseType::Subscribe(serde_json::json!({
+                    "Addrs": ["/ip4/16.3.0.3/tcp/45369"],
+                    "ID": "QmbSLMEMackm7vHiUGMB2EFAPbzeJNpeB9yTpzYKoojDWc"
+                }))
+            },
+            response
+        );
+    }
 }