* Handle error as StatusRuntimeException

* Add test with error from server
* Fix typos in the doc

Author: artembilan

spring-integration-grpc/src/main/java/org/springframework/integration/grpc/inbound/GrpcInboundGateway.java

@@ -22,6 +22,8 @@
 import io.grpc.BindableService;
 import io.grpc.MethodDescriptor;
 import io.grpc.ServerServiceDefinition;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
@@ -152,15 +154,19 @@ private void unary(MethodDescriptor<?, ?> methodDescriptor, Object requestPayloa
 			StreamObserver<Object> responseObserver) {
 
 		sendRequestAndProduceReply(methodDescriptor, requestPayload)
-				.subscribe(responseObserver::onNext, responseObserver::onError, responseObserver::onCompleted);
+				.subscribe(responseObserver::onNext,
+						t -> responseObserver.onError(toGrpcStatusException(t)),
+						responseObserver::onCompleted);
 	}
 
 	private void serverStreaming(MethodDescriptor<?, ?> methodDescriptor, Object requestPayload,
 			StreamObserver<Object> responseObserver) {
 
 		sendRequestAndProduceReply(methodDescriptor, requestPayload)
 				.flatMapMany(payload -> payload instanceof Flux<?> flux ? flux : Flux.just(payload))
-				.subscribe(responseObserver::onNext, responseObserver::onError, responseObserver::onCompleted);
+				.subscribe(responseObserver::onNext,
+						t -> responseObserver.onError(toGrpcStatusException(t)),
+						responseObserver::onCompleted);
 	}
 
 	private StreamObserver<?> clientStreaming(MethodDescriptor<?, ?> methodDescriptor,
@@ -177,15 +183,15 @@ public void onNext(Object value) {
 
 			@Override
 			public void onError(Throwable t) {
-				throw new IllegalStateException(
-						"gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed", t);
+				throw toGrpcStatusException(t, "gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed");
 			}
 
 			@Override
 			public void onCompleted() {
 				requestPayload.tryEmitComplete();
 				sendRequestAndProduceReply(methodDescriptor, requestPayload.asFlux())
-						.subscribe(responseObserver::onNext, responseObserver::onError,
+						.subscribe(responseObserver::onNext,
+								t -> responseObserver.onError(toGrpcStatusException(t)),
 								responseObserver::onCompleted);
 			}
 
@@ -200,13 +206,13 @@ private StreamObserver<?> bidiStreaming(MethodDescriptor<?, ?> methodDescriptor,
 			@Override
 			public void onNext(Object value) {
 				sendRequestAndProduceReply(methodDescriptor, value)
-						.subscribe(responseObserver::onNext, responseObserver::onError);
+						.subscribe(responseObserver::onNext,
+								t -> responseObserver.onError(toGrpcStatusException(t)));
 			}
 
 			@Override
 			public void onError(Throwable t) {
-				throw new IllegalStateException(
-						"gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed", t);
+				throw toGrpcStatusException(t, "gRPC request [" + methodDescriptor.getFullMethodName() + "] has failed");
 			}
 
 			@Override
@@ -231,4 +237,14 @@ private Mono<?> sendRequestAndProduceReply(MethodDescriptor<?, ?> serviceMethod,
 				.map(Message::getPayload);
 	}
 
+	private static StatusRuntimeException toGrpcStatusException(Throwable throwable) {
+		return toGrpcStatusException(throwable, throwable.getMessage());
+	}
+
+	private static StatusRuntimeException toGrpcStatusException(Throwable throwable, @Nullable String description) {
+		return Status.fromThrowable(throwable)
+				.withDescription(description)
+				.asRuntimeException();
+	}
+
 }

spring-integration-grpc/src/test/java/org/springframework/integration/grpc/TestInProcessConfiguration.java

@@ -73,7 +73,7 @@ void startServer() throws IOException {
 
 	@Override
 	public void destroy() {
-		this.server.shutdown();
+		this.server.shutdownNow();
 	}
 
 }

spring-integration-grpc/src/test/java/org/springframework/integration/grpc/inbound/GrpcInboundGatewayTests.java

@@ -27,6 +27,8 @@
 
 import com.google.common.util.concurrent.ListenableFuture;
 import io.grpc.ManagedChannel;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
@@ -49,6 +51,7 @@
 import org.springframework.util.StringUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 /**
  * @author Artem Bilan
@@ -146,6 +149,18 @@ void bidiStreaming() throws InterruptedException {
 				.containsAll(Arrays.stream(names).map("Hello "::concat).toList());
 	}
 
+	@Test
+	void errorFromServer() {
+		assertThatExceptionOfType(StatusRuntimeException.class)
+				.isThrownBy(() -> this.testHelloWorldBlockingStub.errorOnHello(newHelloRequest("Error")))
+				.satisfies(e -> {
+					assertThat(e.getStatus().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
+					assertThat(e.getStatus().getDescription())
+							.contains("Failed to transform Message in bean " +
+									"'grpcIntegrationFlow.subFlow#4.method-invoking-transformer#1'");
+				});
+	}
+
 	private static HelloRequest newHelloRequest(String message) {
 		return HelloRequest.newBuilder().setName(message).build();
 	}
@@ -218,6 +233,12 @@ IntegrationFlow grpcIntegrationFlow(GrpcInboundGateway helloWorldService) {
 
 									.subFlowMapping("BidiStreamHello", flow -> flow
 											.transform(this::requestReply))
+
+									.subFlowMapping("ErrorOnHello", flow -> flow
+											.transform(p -> {
+												throw Status.UNAVAILABLE.withDescription("intentional")
+														.asRuntimeException();
+											}))
 					)
 					.get();
 		}

spring-integration-grpc/src/test/proto/test_hello.proto

@@ -23,6 +23,9 @@ service TestHelloWorld {
   // Streams requests and replies
   rpc BidiStreamHello(stream HelloRequest) returns (stream HelloReply) {}
 
+  // Fail with error
+  rpc ErrorOnHello(HelloRequest) returns (HelloReply) {}
+
 }
 
 // The request message containing the user's name.

src/reference/antora/modules/ROOT/pages/grpc.adoc

@@ -66,7 +66,7 @@ The `GrpcInboundGateway` is a `MessagingGatewaySupport` implementation to receiv
 For initialization, the instance of this gateway requires only an abstract gRPC service class implementing `BindableService`, usually generated from Protobuf and comes with a `*ImplBase` class name.
 
 WARNING: Only standard gRPC services are supported: a generated `AsyncService` contract is what `GrpcInboundGateway` logic based on.
-The Reactor and Kotlin-based service generation don't make sense in Spring Integration logic since those types are not exposed anyhow from the gateway definition.
+The Reactor and Kotlin-based service generation don't make sense in Spring Integration logic since those types are not exposed from the gateway definition.
 
 The gateway uses the mentioned `AsyncService` interface to create proxy and intercept gRPC service methods.
 
@@ -81,17 +81,17 @@ GrpcInboundGateway helloWorldService() {
 ----
 
 The `GrpcInboundGateway` implements a `BindableService` and exposes a `ServerServiceDefinition` based on the mentioned proxy for an `AsyncService` contract of the gRPC service.
-Therefore, exactly instance of this gateway has to be registered into a `ServerBuilder` and no need in any other `*ImplBase` implementations in the application.
+Therefore, an instance of this gateway has to be registered into a `ServerBuilder` and no need in any other `*ImplBase` implementations in the application.
 
 IMPORTANT: With https://spring.io/projects/spring-grpc[Spring gRPC] and its auto-discovery for `BindableService` implementations, the `GrpcInboundGateway` has to be declared as a top-level bean.
 Therefore, Java DSL API like `IntegrationFlow.from(new GrpcInboundGateway(TestHelloWorldGrpc.TestHelloWorldImplBase.class))` is not recommended because such a `BindableService` implementation won't make it visible for respective Spring gRPC infrastructure.
 
 The `GrpcInboundGateway` uses a `sendAndReceiveMessageReactive()` API to interact with the downstream flow and adapts a `Mono` reply to the gRPC `StreamObserver`.
 As mentioned before, the request message payload is exactly a gRPC request message, and it expects a reply in a from of gRPC response message.
-The downstream logic could be type-safe and deal with gRPC messages similar way as if `*ImplBase` would be implemented manually.
+The downstream logic can be type-safe and deal with gRPC messages similar way as if `*ImplBase` would be implemented manually.
 
 The `MethodDescriptor.MethodType.UNARY` and `MethodDescriptor.MethodType.BIDI_STREAMING` are same from the downstream handling logic perspective.
-In other words, the `BIDI_STREAMING` is handled as a loop on request items and the gateway produces a response item for each of them.
+In other words, the `BIDI_STREAMING` is handled as a loop on request items and the gateway produces a response item immediately into the response `StreamObserver`.
 For different `BIDI_STREAMING` logic, the regular gRPC service implementation is recommended.
 
 The `MethodDescriptor.MethodType.CLIENT_STREAMING` mode produces a message with a `Flux` as a payload of gRPC request items.