ServerStreamingMethodHandler extends AbstractMethodHandler<REQ, RESP, CONTEXT> implements ServerCalls.ServerStreamingMethod<REQ, RESP> { @VisibleForTesting static <RESP> Disposable internalHandleResult(Publisher<RESP> result, StreamObserver<RESP> responseObserver) { AtomicBoolean cancelled = new AtomicBoolean(); AtomicReference<Disposable> disposableRef = new AtomicReference<>(); Disposable disposable = Flux.from(result).subscribe( value -> { if (cancelled.get()) { ReactorExt.safeDispose(disposableRef.get()); return; } try { responseObserver.onNext(value); } catch (Exception e) { cancelled.set(true); logger.warn("Subscriber threw error in onNext handler. Retrying with onError", e); try { responseObserver.onError(e); } catch (Exception e2) { logger.warn("Subscriber threw error in onError handler", e2); } ReactorExt.safeDispose(disposableRef.get()); } }, e -> { if (cancelled.get()) { return; } try { responseObserver.onError(e); } catch (Exception e2) { logger.warn("Subscriber threw error in onError handler", e2); } }, () -> { if (cancelled.get()) { return; } try { responseObserver.onCompleted(); } catch (Exception e) { logger.warn("Subscriber threw error in onCompleted handler", e); } } ); disposableRef.set(disposable); if (cancelled.get()) { ReactorExt.safeDispose(disposable); } return disposable; } ServerStreamingMethodHandler(GrpcToReactorMethodBinding<REQ, RESP> binding,
                                 Supplier<CONTEXT> contextResolver,
                                 Object reactorService); @Override void invoke(REQ request, StreamObserver<RESP> responseObserver);  }
@Test public void testOnNextExceptionHandlerOnSubscribe() { ReplayProcessor<String> publisher = ReplayProcessor.create(2); publisher.onNext("a"); publisher.onNext("b"); Disposable disposable = ServerStreamingMethodHandler.internalHandleResult(publisher, responseObserver); assertThat(disposable.isDisposed()).isTrue(); }
@Test public void testOnNextExceptionHandlerAfterSubscribe() { DirectProcessor<String> publisher = DirectProcessor.create(); Disposable disposable = ServerStreamingMethodHandler.internalHandleResult(publisher, responseObserver); publisher.onNext("a"); publisher.onNext("b"); assertThat(disposable.isDisposed()).isTrue(); }