Return a blocking stream instead of an iterable

Author: vietj

vertx-grpc-docs/src/main/java/examples/GreeterClient.java

@@ -13,6 +13,7 @@
 import io.vertx.grpc.common.ServiceMethod;
 import io.vertx.grpc.common.GrpcMessageDecoder;
 import io.vertx.grpc.common.GrpcMessageEncoder;
+import java.util.stream.Stream;
 
 /**
  * <p>A client for invoking the Greeter gRPC service.</p>

vertx-grpc-docs/src/main/java/examples/StreamingClient.java

@@ -13,6 +13,7 @@
 import io.vertx.grpc.common.ServiceMethod;
 import io.vertx.grpc.common.GrpcMessageDecoder;
 import io.vertx.grpc.common.GrpcMessageEncoder;
+import java.util.stream.Stream;
 
 /**
  * <p>A client for invoking the Streaming gRPC service.</p>
@@ -117,7 +118,7 @@ static StreamingClient create(GrpcClient client, SocketAddress host, io.vertx.gr
   Future<ReadStream<examples.Item>> source(examples.Empty request);
 
   @io.vertx.codegen.annotations.GenIgnore
-  Iterable<examples.Item> source_sync(examples.Empty request);
+  Stream<examples.Item> source_sync(examples.Empty request);
 
   /**
    * Calls the Sink RPC service method.
@@ -208,28 +209,19 @@ public Future<ReadStream<examples.Item>> source(examples.Empty request) {
     });
   }
 
-  public Iterable<examples.Item> source_sync(examples.Empty request) {
-    java.util.Iterator<examples.Item> iterator = source_(request)
+  public Stream<examples.Item> source_sync(examples.Empty request) {
+    Stream<examples.Item> iterator = source_(request)
       .compose(req -> {
         req.end(request);
         return req.response().compose(resp -> {
           if (resp.status() != null && resp.status() != GrpcStatus.OK) {
             return Future.failedFuture("Invalid gRPC status " + resp.status());
           } else {
-            return Future.succeededFuture(io.vertx.core.internal.streams.ReadStreamIterator.iterator(resp));
+            return Future.succeededFuture(resp.blockingStream());
           }
         });
       }).await();
-    return new Iterable<>() {
-      boolean consumed = false;
-      public java.util.Iterator<examples.Item> iterator() {
-        if (consumed) {
-          throw new IllegalStateException();
-        }
-        consumed = true;
-        return iterator;
-      }
-    };
+    return iterator;
   }
 
   public Future<GrpcClientRequest<examples.Empty, examples.Item>> source_(examples.Empty request) {

vertx-grpc-it/src/test/java/io/vertx/grpc/it/DeadlineTest.java

@@ -16,7 +16,9 @@
 import io.grpc.examples.helloworld.HelloRequest;
 import io.vertx.core.Future;
 import io.vertx.core.http.HttpServer;
+import io.vertx.core.internal.streams.ReadStreamIterator;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.core.streams.ReadStream;
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.grpc.client.GrpcClient;
@@ -28,7 +30,9 @@
 import io.vertx.grpc.server.GrpcServerOptions;
 import org.junit.Test;
 
+import java.util.Spliterators;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.StreamSupport;
 
 import static io.grpc.examples.helloworld.GreeterService.SayHello;
 
@@ -39,6 +43,11 @@ public class DeadlineTest extends ProxyTestBase {
 
   @Test
   public void testAutomaticPropagation(TestContext should) {
+
+    ReadStream<String> rs = null;
+
+    StreamSupport.stream(() -> Spliterators.spliteratorUnknownSize(ReadStreamIterator.iterator(rs), 0), 0, false);
+
     Async latch = should.async(3);
     GrpcClientOptions clientOptions = new GrpcClientOptions().setScheduleDeadlineAutomatically(true);
     GrpcServerOptions serverOptions = new GrpcServerOptions().setDeadlinePropagation(true);

vertx-grpc-it/src/test/java/io/vertx/grpc/it/ProtocPluginTest.java

@@ -38,6 +38,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 
 import static io.grpc.examples.helloworld.GreeterService.SayHello;
 import static io.grpc.testing.integration.TestServiceService.*;
@@ -407,12 +408,10 @@ public ReadStream<Messages.StreamingOutputCallResponse> streamingOutputCall(Mess
     Messages.StreamingOutputCallRequest request = Messages.StreamingOutputCallRequest.newBuilder()
       .setPayload(Messages.Payload.newBuilder().setBody(ByteString.copyFrom("StreamingOutputRequest", StandardCharsets.UTF_8)).build())
       .build();
-    Iterable<Messages.StreamingOutputCallResponse> res = client.streamingOutputCall_sync(request);
+    Stream<Messages.StreamingOutputCallResponse> res = client.streamingOutputCall_sync(request);
     Thread.sleep(100);
     List<Messages.StreamingOutputCallResponse> list = new ArrayList<>();
-    for (Messages.StreamingOutputCallResponse msg : res) {
-      list.add(msg);
-    }
+    res.forEach(msg -> list.add(msg));
     assertEquals(2, list.size());
   }
 

vertx-grpc-protoc-plugin2/src/main/resources/client.mustache

@@ -15,6 +15,7 @@ import io.vertx.grpc.common.ServiceName;
 import io.vertx.grpc.common.ServiceMethod;
 import io.vertx.grpc.common.GrpcMessageDecoder;
 import io.vertx.grpc.common.GrpcMessageEncoder;
+import java.util.stream.Stream;
 
 /**
  * <p>A client for invoking the {{serviceName}} gRPC service.</p>
@@ -100,7 +101,7 @@ public interface {{className}} {
   Future<ReadStream<{{outputType}}>> {{vertxMethodName}}({{inputType}} request);
 
   @io.vertx.codegen.annotations.GenIgnore
-  Iterable<{{outputType}}> {{vertxMethodName}}_sync({{inputType}} request);
+  Stream<{{outputType}}> {{vertxMethodName}}_sync({{inputType}} request);
 {{/unaryManyMethods}}
 {{#manyUnaryMethods}}
 
@@ -221,28 +222,19 @@ class {{className}}Impl implements {{className}} {
     });
   }
 
-  public Iterable<{{outputType}}> {{vertxMethodName}}_sync({{inputType}} request) {
-    java.util.Iterator<{{outputType}}> iterator = {{vertxMethodName}}_(request)
+  public Stream<{{outputType}}> {{vertxMethodName}}_sync({{inputType}} request) {
+    Stream<{{outputType}}> iterator = {{vertxMethodName}}_(request)
       .compose(req -> {
         req.end(request);
         return req.response().compose(resp -> {
           if (resp.status() != null && resp.status() != GrpcStatus.OK) {
             return Future.failedFuture("Invalid gRPC status " + resp.status());
           } else {
-            return Future.succeededFuture(io.vertx.core.internal.streams.ReadStreamIterator.iterator(resp));
+            return Future.succeededFuture(resp.blockingStream());
           }
         });
       }).await();
-    return new Iterable<>() {
-      boolean consumed = false;
-      public java.util.Iterator<{{outputType}}> iterator() {
-        if (consumed) {
-          throw new IllegalStateException();
-        }
-        consumed = true;
-        return iterator;
-      }
-    };
+    return iterator;
   }
 
   public Future<GrpcClientRequest<{{inputType}}, {{outputType}}>> {{vertxMethodName}}_({{inputType}} request) {