@Test public void propagateTransientErrorStreamingCall() { StatusRuntimeException retryableException = new StatusRuntimeException( Status.INTERNAL.withDescription("HTTP/2 error code: INTERNAL_ERROR")); Flux<Void> result = ObservableReactiveUtil.streamingCall(observer -> observer.onError(retryableException)); StepVerifier.create(result) .expectErrorSatisfies(throwable -> assertThat(throwable) .hasCauseInstanceOf(StatusRuntimeException.class) .isInstanceOf(R2dbcTransientResourceException.class)) .verify(); }
public static <RequestT, ResponseT> Flux<ResponseT> streamingCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Flux.create(sink -> { StreamingObserver observer = new StreamingObserver(sink); remoteCall.accept(observer); sink.onRequest(demand -> observer.request(demand)); }); }
ObservableReactiveUtil { public static <RequestT, ResponseT> Flux<ResponseT> streamingCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Flux.create(sink -> { StreamingObserver observer = new StreamingObserver(sink); remoteCall.accept(observer); sink.onRequest(demand -> observer.request(demand)); }); } }
ObservableReactiveUtil { public static <RequestT, ResponseT> Flux<ResponseT> streamingCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Flux.create(sink -> { StreamingObserver observer = new StreamingObserver(sink); remoteCall.accept(observer); sink.onRequest(demand -> observer.request(demand)); }); }  }
ObservableReactiveUtil { public static <RequestT, ResponseT> Flux<ResponseT> streamingCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Flux.create(sink -> { StreamingObserver observer = new StreamingObserver(sink); remoteCall.accept(observer); sink.onRequest(demand -> observer.request(demand)); }); }  static Mono<ResponseT> unaryCall(
      Consumer<StreamObserver<ResponseT>> remoteCall); static Flux<ResponseT> streamingCall(
      Consumer<StreamObserver<ResponseT>> remoteCall); }
ObservableReactiveUtil { public static <RequestT, ResponseT> Flux<ResponseT> streamingCall( Consumer<StreamObserver<ResponseT>> remoteCall) { return Flux.create(sink -> { StreamingObserver observer = new StreamingObserver(sink); remoteCall.accept(observer); sink.onRequest(demand -> observer.request(demand)); }); }  static Mono<ResponseT> unaryCall(
      Consumer<StreamObserver<ResponseT>> remoteCall); static Flux<ResponseT> streamingCall(
      Consumer<StreamObserver<ResponseT>> remoteCall);  }