class 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);

}