Fix: use json payloads

.github/workflows/ci.yml

@@ -114,7 +114,7 @@ jobs:
           testground healthcheck --runner local:docker --fix;
         shell: bash
 
-      - name: Run testground plan (sdk-rust)
+      - name: Run testground plan (case=example)
         run: |
           testground run single       \
             --plan=sdk-rust           \
@@ -124,12 +124,24 @@ jobs:
             --instances=1             \
             --wait                    \
             --collect                 \
-            --collect-file ./result.tgz
+            --collect-file ./result_example.tgz
+
+      - name: Run testground plan (case=publish-subscribe)
+        run: |
+          testground run single          \
+            --plan=sdk-rust              \
+            --testcase=publish-subscribe \
+            --builder=docker:generic     \
+            --runner=local:docker        \
+            --instances=2                \
+            --wait                       \
+            --collect                    \
+            --collect-file ./result_publish_subscribe.tgz
 
       - uses: actions/upload-artifact@v3
         if: ${{ always() }}
         with:
           name: testground-output
           path: |
             testground/daemon.*
-            result.tgz
+            result*.tgz

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,7 +1,19 @@
+use std::borrow::Cow;
+
+use tokio_stream::StreamExt;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let client = testground::client::Client::new_and_init().await?;
 
+    match client.run_parameters().test_case.as_str() {
+        "example" => example(client).await,
+        "publish-subscribe" => publish_subscribe(client).await,
+        _ => panic!("Unknown test case: {}", client.run_parameters().test_case),
+    }
+}
+
+async fn example(client: testground::client::Client) -> Result<(), Box<dyn std::error::Error>> {
     client.record_message(format!(
         "{}, sdk-rust!",
         client
@@ -11,7 +23,51 @@ 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(())
 }
+
+async fn publish_subscribe(
+    client: testground::client::Client,
+) -> Result<(), Box<dyn std::error::Error>> {
+    client.record_message("running the publish_subscribe test");
+
+    match client.global_seq() {
+        1 => {
+            client.record_message("I am instance 1: acting as the leader");
+
+            let json = serde_json::json!({"foo": "bar"});
+            client.publish("demonstration", Cow::Owned(json)).await?;
+            client.record_success().await?;
+        }
+        _ => {
+            client.record_message(format!(
+                "I am instance {}: acting as a follower",
+                client.global_seq()
+            ));
+
+            let payload = client
+                .subscribe("demonstration")
+                .await
+                .take(1)
+                .map(|x| x.unwrap())
+                .next()
+                .await
+                .unwrap();
+
+            client.record_message(format!("I received the payload: {}", payload));
+
+            if payload["foo"].as_str() == Some("bar") {
+                client.record_success().await?;
+            } else {
+                client
+                    .record_failure(format!("invalid payload: {}", payload))
+                    .await?;
+            }
+        }
+    }
+    Ok(())
+}

manifest.toml

@@ -15,4 +15,8 @@ name = "example"
 instances = { min = 1, max = 1, default = 1 }
 
   [testcases.params]
-  greeting = { type = "string", desc = "greeting", default = "Hello" }
+  greeting = { type = "string", desc = "greeting", default = "Hello" }
+
+[[testcases]]
+name = "publish-subscribe"
+instances = { min = 2, max = 10, default = 2 }

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

@@ -30,7 +30,7 @@ pub struct RawResponse {
 pub enum ResponseType {
     SignalEntry { seq: u64 },
     Publish { seq: u64 },
-    Subscribe(String),
+    Subscribe(serde_json::Value),
     Error(String),
     Barrier,
 }
@@ -54,14 +54,14 @@ impl From<RawResponse> for Response {
         let response = match (error, subscribe, signal_entry, publish) {
             (None, None, None, None) => ResponseType::Barrier,
             (Some(error), None, None, None) => {
-                //Hack to remove extra escape characters
+                // Hack to remove extra escape characters
                 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)
+                // the Subscribe payload is a json encoded string, so we need to deserialize it
+                let payload = serde_json::from_str(&msg).expect("JSON Deserialization");
+                ResponseType::Subscribe(payload)
             }
             (None, None, Some(signal), None) => ResponseType::SignalEntry { seq: signal.seq },
             (None, None, None, Some(publish)) => ResponseType::Publish { seq: publish.seq },
@@ -89,7 +89,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 +104,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
+        );
+    }
 }