Reproducible failure, run the client with `cargo run --bin blob-clien…

…t 10000 10000`

tonic-examples/Cargo.toml

@@ -44,6 +44,14 @@ path = "src/tls/client.rs"
 name = "tls-server"
 path = "src/tls/server.rs"
 
+[[bin]]
+name = "blob-client"
+path = "src/blobservice/client.rs"
+
+[[bin]]
+name = "blob-server"
+path = "src/blobservice/server.rs"
+
 [dependencies]
 tonic = { path = "../tonic", features = ["rustls"] }
 bytes = "0.4"

tonic-examples/build.rs

@@ -2,4 +2,5 @@ fn main() {
     tonic_build::compile_protos("proto/helloworld/helloworld.proto").unwrap();
     tonic_build::compile_protos("proto/routeguide/route_guide.proto").unwrap();
     tonic_build::compile_protos("proto/echo/echo.proto").unwrap();
+    tonic_build::compile_protos("proto/blobservice/blobservice.proto").unwrap();
 }

tonic-examples/proto/blobservice/blobservice.proto

@@ -0,0 +1,30 @@
+// Copyright 2015 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package blobservice;
+
+service Blobber {
+  rpc GetBytes (BlobRequest) returns (stream BlobResponse) {}
+}
+
+message BlobRequest {
+  uint64 nbytes = 1;
+  uint64 nelements = 2;
+}
+
+message BlobResponse {
+  bytes bytes = 1;
+}

tonic-examples/src/blobservice/client.rs

@@ -0,0 +1,27 @@
+use futures::TryStreamExt;
+pub mod blobservice {
+    tonic::include_proto!("blobservice");
+}
+
+use blobservice::{client::BlobberClient, BlobRequest};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let mut client = BlobberClient::connect("http://[::1]:50051")?;
+    let nelements = std::env::args().nth(1).unwrap().parse::<u64>().unwrap();
+    let nbytes = std::env::args().nth(2).unwrap().parse::<u64>().unwrap();
+
+    let request = tonic::Request::new(BlobRequest { nelements, nbytes });
+
+    let response = client.get_bytes(request).await?;
+    let mut inner = response.into_inner();
+    let mut i = 0u64;
+    while let Some(_) = inner.try_next().await? {
+        if i % (nelements as f64 / 10.0) as u64 == 0 {
+            println!("i = {:?}", i);
+        }
+        i += 1;
+    }
+
+    Ok(())
+}

tonic-examples/src/blobservice/server.rs

@@ -0,0 +1,50 @@
+use futures::Stream;
+use std::pin::Pin;
+use tonic::{transport::Server, Request, Response, Status};
+
+pub mod blobservice {
+    tonic::include_proto!("blobservice");
+}
+
+use blobservice::{
+    server::{Blobber, BlobberServer},
+    BlobRequest, BlobResponse,
+};
+
+#[derive(Default)]
+pub struct SimpleBlobber;
+
+#[tonic::async_trait]
+impl Blobber for SimpleBlobber {
+    type GetBytesStream =
+        Pin<Box<dyn Stream<Item = Result<BlobResponse, Status>> + Send + 'static>>;
+
+    async fn get_bytes(
+        &self,
+        request: Request<BlobRequest>,
+    ) -> Result<Response<Self::GetBytesStream>, Status> {
+        let inner = request.into_inner();
+        let nbytes = inner.nbytes;
+        let nelements = inner.nelements;
+        let bytes = (0..nbytes).map(|value| value as u8).collect::<Vec<_>>();
+        let reply = futures::stream::iter((0..nelements).map(move |_| {
+            Ok(blobservice::BlobResponse {
+                bytes: bytes.clone(),
+            })
+        }));
+
+        Ok(Response::new(Box::pin(reply) as Self::GetBytesStream))
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:50051".parse().unwrap();
+    let greeter = SimpleBlobber::default();
+
+    Server::builder()
+        .serve(addr, BlobberServer::new(greeter))
+        .await?;
+
+    Ok(())
+}