Add streaming method to service

Signed-off-by: Dave Syer <[email protected]>

Author: dsyer

spring-cloud-gateway-integration-tests/grpc/src/main/java/org/springframework/cloud/gateway/tests/grpc/GRPCApplication.java

@@ -73,7 +73,11 @@ private void start() throws IOException {
 			Integer serverPort = environment.getProperty("local.server.port", Integer.class);
 			int grpcPort = serverPort + 1;
 			ServerCredentials creds = createServerCredentials();
-			server = Grpc.newServerBuilderForPort(grpcPort, creds).addService(new HelloService()).build().start();
+			server = Grpc.newServerBuilderForPort(grpcPort, creds)
+				.addService(new HelloService())
+				.addService(new StreamService())
+				.build()
+				.start();
 
 			log.info("Starting gRPC server in port " + grpcPort);
 
@@ -101,6 +105,38 @@ private void stop() throws InterruptedException {
 			log.info("gRPC server stopped");
 		}
 
+		static class StreamService extends StreamServiceGrpc.StreamServiceImplBase {
+
+			@Override
+			public void more(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
+				int count = 0;
+				while (count < 3) {
+					HelloResponse reply = HelloResponse.newBuilder()
+						.setGreeting("Hello(" + count + ") ==> " + request.getFirstName())
+						.build();
+					if ("failWithRuntimeExceptionAfterData!".equals(request.getFirstName()) && count == 2) {
+						StatusRuntimeException exception = Status.RESOURCE_EXHAUSTED
+							.withDescription("Too long firstNames?")
+							.asRuntimeException();
+						responseObserver.onError(exception);
+						return;
+					}
+					responseObserver.onNext(reply);
+					count++;
+					try {
+						Thread.sleep(200L);
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+						responseObserver.onError(e);
+						return;
+					}
+				}
+				responseObserver.onCompleted();
+			}
+
+		}
+
 		static class HelloService extends HelloServiceGrpc.HelloServiceImplBase {
 
 			@Override
@@ -122,7 +158,7 @@ public void hello(HelloRequest request, StreamObserver<HelloResponse> responseOb
 
 				if ("failWithRuntimeExceptionAfterData!".equals(request.getFirstName())) {
 					StatusRuntimeException exception = Status.RESOURCE_EXHAUSTED.withDescription("Too long firstNames?")
-							.asRuntimeException();
+						.asRuntimeException();
 					responseObserver.onError(exception);
 					return;
 				}

spring-cloud-gateway-integration-tests/grpc/src/main/proto/hello.pb

@@ -0,0 +1,10 @@
+syntax = "proto3";
+option java_multiple_files = true;
+option java_package = "org.springframework.cloud.gateway.tests.grpc";
+package org.springframework.cloud.gateway.tests.grpc;
+
+import "hello.proto";
+
+service StreamService {
+  rpc more(HelloRequest) returns (stream HelloResponse);
+}

spring-cloud-gateway-integration-tests/grpc/src/main/proto/stream.proto

@@ -17,6 +17,7 @@
 package org.springframework.cloud.gateway.tests.grpc;
 
 import java.security.cert.X509Certificate;
+import java.util.Iterator;
 
 import javax.net.ssl.SSLException;
 import javax.net.ssl.TrustManager;
@@ -72,6 +73,18 @@ public void gRPCUnaryCallShouldReturnResponse() throws SSLException {
 		Assertions.assertThat(response.getGreeting()).isEqualTo("Hello, Sir FromClient");
 	}
 
+	@Test
+	public void gRPCStreamingCallShouldReturnResponse() throws SSLException {
+		ManagedChannel channel = createSecuredChannel(gatewayPort);
+
+		final Iterator<HelloResponse> response = StreamServiceGrpc.newBlockingStub(channel)
+			.more(HelloRequest.newBuilder().setFirstName("Sir").setLastName("FromClient").build());
+
+		Assertions.assertThat(response.next().getGreeting()).isEqualTo("Hello(0) ==> Sir");
+		Assertions.assertThat(response.next().getGreeting()).isEqualTo("Hello(1) ==> Sir");
+		Assertions.assertThat(response.next().getGreeting()).isEqualTo("Hello(2) ==> Sir");
+	}
+
 	private ManagedChannel createSecuredChannel(int port) throws SSLException {
 		TrustManager[] trustAllCerts = createTrustAllTrustManager();
 
@@ -97,13 +110,38 @@ public void gRPCUnaryCallShouldHandleRuntimeException() throws SSLException {
 	}
 
 	@Test
-	public void gRPCUnaryCallShouldHandleRuntimeException2() throws SSLException {
+	public void gRPCUnaryCallShouldHandleRuntimeExceptionAfterData() throws SSLException {
 		ManagedChannel channel = createSecuredChannel(gatewayPort);
 		boolean thrown = false;
 		try {
 			HelloServiceGrpc.newBlockingStub(channel)
-					.hello(HelloRequest.newBuilder().setFirstName("failWithRuntimeExceptionAfterData!").build())
-					.getGreeting();
+				.hello(HelloRequest.newBuilder().setFirstName("failWithRuntimeExceptionAfterData!").build())
+				.getGreeting();
+		}
+		catch (StatusRuntimeException e) {
+			thrown = true;
+			Assertions.assertThat(e.getStatus().getCode()).isEqualTo(RESOURCE_EXHAUSTED.getCode());
+			Assertions.assertThat(e.getStatus().getDescription()).isEqualTo("Too long firstNames?");
+		}
+		Assertions.assertThat(thrown).withFailMessage("Expected exception not thrown!").isTrue();
+	}
+
+	@Test
+	public void gRPCStreamingCallShouldHandleRuntimeExceptionAfterData() throws SSLException {
+		ManagedChannel channel = createSecuredChannel(gatewayPort);
+		boolean thrown = false;
+		final Iterator<HelloResponse> response = StreamServiceGrpc.newBlockingStub(channel)
+			.more(HelloRequest.newBuilder()
+				.setFirstName("failWithRuntimeExceptionAfterData!")
+				.setLastName("FromClient")
+				.build());
+		Assertions.assertThat(response.next().getGreeting())
+			.isEqualTo("Hello(0) ==> failWithRuntimeExceptionAfterData!");
+		Assertions.assertThat(response.next().getGreeting())
+			.isEqualTo("Hello(1) ==> failWithRuntimeExceptionAfterData!");
+		try {
+			Assertions.assertThat(response.next().getGreeting())
+				.isEqualTo("Hello(2) ==> failWithRuntimeExceptionAfterData!");
 		}
 		catch (StatusRuntimeException e) {
 			thrown = true;