test: introduce publish-subscribe example & test

.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=1                \
+            --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

examples/example.rs

@@ -1,9 +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
@@ -14,10 +24,47 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     ));
 
     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 }