Existing gRPC exceptions can be passed through unchanged

They aren't particularly exceptional and there is no need to wrap
them again.

Author: Dave Syer

src/main/java/io/projectriff/invoker/ExceptionConverter.java

@@ -6,12 +6,20 @@
 import java.util.concurrent.TimeoutException;
 
 import io.grpc.Status;
+import io.grpc.StatusException;
+import io.grpc.StatusRuntimeException;
 
 /**
  * Utility class for converting exceptions to gRPC status values.
  */
 final class ExceptionConverter {
 	static Status createStatus(Throwable t) {
+		if (t instanceof StatusRuntimeException) {
+			return ((StatusRuntimeException) t).getStatus();
+		}
+		if (t instanceof StatusException) {
+			return ((StatusException) t).getStatus();
+		}
 		return Status.fromCode(determineCode(t)).withDescription(t.getMessage())
 				.withCause(t);
 	}

src/main/java/io/projectriff/invoker/JavaFunctionInvokerServer.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import io.projectriff.grpc.function.MessageFunctionGrpc;
 
@@ -119,9 +120,16 @@ public StreamObserver<io.projectriff.grpc.function.FunctionProtos.Message> call(
 				message -> responseObserver
 						.onNext(MessageConversionUtils.toGrpc(payloadToBytes(message))),
 				t -> {
+					if (t instanceof StatusRuntimeException) {
+						// Not particularly exceptional.
+						logger.info("RPC ending: " + t.getClass().getName() + " ("
+								+ t.getMessage() + ")");
+					}
+					else {
+						t = ExceptionConverter.createStatus(t).asRuntimeException();
+						logger.error("RPC ending", t);
+					}
 					responseObserver.onError(t);
-					throw new IllegalStateException(
-							ExceptionConverter.createStatus(t).asException());
 				}, () -> {
 					// Make sure the emitter is disposed (should work even if it already
 					// was since it's idempotent)

src/test/java/io/projectriff/invoker/ExceptionConverterTests.java

@@ -6,6 +6,8 @@
 import java.util.concurrent.TimeoutException;
 
 import io.grpc.Status;
+import io.grpc.StatusException;
+import io.grpc.StatusRuntimeException;
 
 import org.junit.Test;
 
@@ -15,6 +17,17 @@ public class ExceptionConverterTests {
 
 	private static final String TEST_MESSAGE = "test message";
 
+	@Test
+	public void statusRuntime() {
+		checkStatus(new StatusRuntimeException(Status.CANCELLED), Status.Code.CANCELLED,
+				null);
+	}
+
+	@Test
+	public void status() {
+		checkStatus(new StatusException(Status.CANCELLED), Status.Code.CANCELLED, null);
+	}
+
 	@Test
 	public void cancelled() {
 		checkStatus(new CancellationException(TEST_MESSAGE), Status.Code.CANCELLED);
@@ -107,10 +120,17 @@ public void unauthenticated() {
 	}
 
 	private void checkStatus(Throwable cause, Status.Code expectedCode) {
+		checkStatus(cause, expectedCode, TEST_MESSAGE);
+	}
+
+	private void checkStatus(Throwable cause, Status.Code expectedCode,
+			String description) {
 		Status s = ExceptionConverter.createStatus(cause);
 		assertThat(s.getCode()).isEqualTo(expectedCode);
-		assertThat(s.getDescription()).isEqualTo(TEST_MESSAGE);
-		assertThat(s.getCause()).isEqualTo(cause);
+		assertThat(s.getDescription()).isEqualTo(description);
+		if (s.getCause() != null) {
+			assertThat(s.getCause()).isEqualTo(cause);
+		}
 	}
 
 	@SuppressWarnings("serial")