@Override
    public StreamObserver<StreamingOutputCallRequest> fullDuplexCall(
            final StreamObserver<StreamingOutputCallResponse> responseObserver) {
        final ResponseDispatcher dispatcher = new ResponseDispatcher(responseObserver);
        return new StreamObserver<StreamingOutputCallRequest>() {
            @Override
            public void onNext(StreamingOutputCallRequest request) {
                if (request.hasResponseStatus()) {
                    dispatcher.cancel();
                    responseObserver.onError(Status.fromCodeValue(request.getResponseStatus().getCode())
                                                   .withDescription(
                                                           request.getResponseStatus().getMessage())
                                                   .asRuntimeException());
                    return;
                }
                dispatcher.enqueue(toChunkQueue(request));
            }

            @Override
            public void onCompleted() {
                if (!dispatcher.isCancelled()) {
                    // Tell the dispatcher that all input has been received.
                    dispatcher.completeInput();
                }
            }

            @Override
            public void onError(Throwable cause) {
                responseObserver.onError(cause);
            }
        };
    }
True bi-directional streaming. Processes requests as they come in. Begins streaming results
immediately.
grpc/src/test/java/com/linecorp/armeria/internal/common/grpc/TestServiceImpl.java