Merge pull request containerd#272 from teawater/exp_server_send

examples: Add example for server_send_stream

Author: lifupan

example/async-stream-client.rs

@@ -44,9 +44,12 @@ async fn main() {
     let sc1 = sc.clone();
     let t6 = tokio::spawn(echo_null_stream(sc1));
 
-    let t7 = tokio::spawn(echo_default_value(sc));
+    let sc1 = sc.clone();
+    let t7 = tokio::spawn(echo_default_value(sc1));
+
+    let t8 = tokio::spawn(server_send_stream(sc));
 
-    let _ = tokio::join!(t1, t2, t3, t4, t5, t6, t7);
+    let _ = tokio::join!(t1, t2, t3, t4, t5, t6, t7, t8);
 }
 
 fn default_ctx() -> Context {
@@ -201,3 +204,18 @@ async fn echo_default_value(cli: streaming_ttrpc::StreamingClient) {
     assert_eq!(received.seq, 0);
     assert_eq!(received.msg, "");
 }
+
+#[cfg(unix)]
+async fn server_send_stream(cli: streaming_ttrpc::StreamingClient) {
+    let mut stream = cli
+        .server_send_stream(default_ctx(), &Default::default())
+        .await
+        .unwrap();
+
+    let mut seq = 0;
+    while let Some(received) = stream.recv().await.unwrap() {
+        assert_eq!(received.seq, seq);
+        assert_eq!(received.msg, "hello");
+        seq += 1;
+    }
+}

example/async-stream-server.rs

@@ -152,6 +152,25 @@ impl streaming_ttrpc::Streaming for StreamingService {
 
         Ok(())
     }
+
+    async fn server_send_stream(
+        &self,
+        _ctx: &::ttrpc::r#async::TtrpcContext,
+        _: empty::Empty,
+        s: ::ttrpc::r#async::ServerStreamSender<streaming::EchoPayload>,
+    ) -> ::ttrpc::Result<()> {
+        let mut seq = 0;
+        while seq < 10 {
+            sleep(std::time::Duration::from_millis(100)).await;
+            let mut e = streaming::EchoPayload::new();
+            e.seq = seq;
+            e.msg = format!("hello");
+            s.send(&e).await.unwrap();
+            seq += 1;
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(windows)]

example/protocols/protos/streaming.proto

@@ -33,6 +33,7 @@ service Streaming {
 	rpc EchoNull(stream EchoPayload) returns (google.protobuf.Empty);
 	rpc EchoNullStream(stream EchoPayload) returns (stream google.protobuf.Empty);
 	rpc EchoDefaultValue(EchoPayload) returns (stream EchoPayload);
+	rpc ServerSendStream(google.protobuf.Empty) returns (stream EchoPayload);
 }
 
 message EchoPayload {