### Question:
ObservableReactiveUtil { public static <ResponseT> Mono<ResponseT> unaryCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Mono.create(sink -> remoteCall.accept(new UnaryStreamObserver(sink))); }  static Mono<ResponseT> unaryCall(
      Consumer<StreamObserver<ResponseT>> remoteCall); static Flux<ResponseT> streamingCall(
      Consumer<StreamObserver<ResponseT>> remoteCall);  }### Answer:
@Test public void unaryCallForwardsError() { Mono<Integer> mono = ObservableReactiveUtil.unaryCall(observer -> { observer.onError(new IllegalArgumentException("oh no")); }); StepVerifier.create(mono) .expectErrorMatches(throwable -> throwable instanceof R2dbcNonTransientResourceException && throwable.getMessage().equals("oh no")) .verify(); }
@Test public void unaryCallThrowsExceptionIfCompletedWithNoValue() { Mono<Integer> mono = ObservableReactiveUtil.unaryCall(observer -> observer.onCompleted()); StepVerifier.create(mono) .expectErrorMatches(throwable -> throwable instanceof RuntimeException && throwable.getMessage().equals( "Unary gRPC call completed without yielding a value or an error")) .verify(); }
@Test public void propagateTransientErrorUnaryCall() { StatusRuntimeException retryableException = new StatusRuntimeException( Status.INTERNAL.withDescription("HTTP/2 error code: INTERNAL_ERROR")); Mono<Void> result = ObservableReactiveUtil.unaryCall(observer -> observer.onError(retryableException)); StepVerifier.create(result) .expectErrorSatisfies(throwable -> assertThat(throwable).hasCauseInstanceOf(StatusRuntimeException.class) .isInstanceOf(R2dbcTransientResourceException.class)) .verify(); }
@Test public void propagateNonRetryableError() { Mono<Void> result = ObservableReactiveUtil.unaryCall( observer -> observer.onError(new IllegalArgumentException())); StepVerifier.create(result) .expectErrorSatisfies(throwable -> assertThat(throwable).hasCauseInstanceOf(IllegalArgumentException.class) .isInstanceOf(R2dbcNonTransientException.class)) .verify(); }
@Test public void unaryCallReturnsSingleValue() { Mono<Integer> mono = ObservableReactiveUtil.unaryCall(observer -> { observer.onNext(42); observer.onCompleted(); }); StepVerifier.create(mono) .expectNext(42) .verifyComplete(); }