servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/ProtocolCompatibilityTest.java
// Wrap grpc client in our client interface to simplify test code
Wrap grpc client in our client interface to simplify test code
Wrap grpc client in our client interface to simplify test code
private static CompatClient grpcJavaClient(final SocketAddress address, @Nullable final String compression,
                                               final boolean ssl,
                                               @Nullable Duration timeout) throws Exception {
        final NettyChannelBuilder builder = NettyChannelBuilder.forAddress(address);
        if (ssl) {
            final SslContext context = GrpcSslContexts.forClient()
                    .trustManager(InsecureTrustManagerFactory.INSTANCE)
                    .build();
            builder.sslContext(context);
        } else {
            builder.usePlaintext();
        }
        final ManagedChannel channel = builder.build();
        CompatGrpc.CompatStub stub = CompatGrpc.newStub(channel);
        if (compression != null) {
            stub = stub.withCompression(compression);
        }
        if (null != timeout) {
            stub = stub.withDeadlineAfter(timeout.toNanos(), NANOSECONDS);
        }
        final CompatGrpc.CompatStub finalStub = stub;
        return new CompatClient() {
            @Override
            public GrpcExecutionContext executionContext() {
                throw new UnsupportedOperationException();
            }
            @Override
            public Publisher<CompatResponse> bidirectionalStreamingCall(final Publisher<CompatRequest> request) {
                final PublisherSource.Processor<CompatResponse, CompatResponse> processor =
                        newPublisherProcessor(3);
                sendRequest(request, finalStub.bidirectionalStreamingCall(adaptResponse(processor)));
                return fromSource(processor);
            }
            @Override
            public Publisher<CompatResponse> bidirectionalStreamingCall(
                    final GrpcClientMetadata metadata, final Publisher<CompatRequest> request) {
                return bidirectionalStreamingCall(request);
            }
            @Deprecated
            @Override
            public Publisher<CompatResponse> bidirectionalStreamingCall(
                    final BidirectionalStreamingCallMetadata metadata, final Publisher<CompatRequest> request) {
                return bidirectionalStreamingCall(request);
            }
            @SuppressWarnings("unchecked")
            @Override
            public Single<CompatResponse> clientStreamingCall(final Publisher<CompatRequest> request) {
                final Processor<CompatResponse, CompatResponse> processor = newSingleProcessor();
                final StreamObserver<CompatRequest> requestObserver =
                        finalStub.clientStreamingCall(adaptResponse(processor));
                sendRequest(request, requestObserver);
                return (Single<CompatResponse>) processor;
            }
            @Deprecated
            @Override
            public Single<CompatResponse> clientStreamingCall(final ClientStreamingCallMetadata metadata,
                                                              final Publisher<CompatRequest> request) {
                return clientStreamingCall(request);
            }
            @Override
            public Single<CompatResponse> clientStreamingCall(final GrpcClientMetadata metadata,
                                                              final Publisher<CompatRequest> request) {
                return clientStreamingCall(request);
            }
            @SuppressWarnings("unchecked")
            @Override
            public Single<CompatResponse> scalarCall(final CompatRequest request) {
                final Processor<CompatResponse, CompatResponse> processor = newSingleProcessor();
                finalStub.scalarCall(request, adaptResponse(processor));
                return (Single<CompatResponse>) processor;
            }
            @Deprecated
            @Override
            public Single<CompatResponse> scalarCall(final ScalarCallMetadata metadata, final CompatRequest request) {
                return scalarCall(request);
            }
            @Override
            public Single<CompatResponse> scalarCall(final GrpcClientMetadata metadata, final CompatRequest request) {
                return scalarCall(request);
            }
            @Override
            public Publisher<CompatResponse> serverStreamingCall(final CompatRequest request) {
                final PublisherSource.Processor<CompatResponse, CompatResponse> processor =
                        newPublisherProcessor(3);
                finalStub.serverStreamingCall(request, adaptResponse(processor));
                return fromSource(processor);
            }
            @Deprecated
            @Override
            public Publisher<CompatResponse> serverStreamingCall(final ServerStreamingCallMetadata metadata,
                                                                 final CompatRequest request) {
                return serverStreamingCall(request);
            }
            @Override
            public Publisher<CompatResponse> serverStreamingCall(final GrpcClientMetadata metadata,
                                                                 final CompatRequest request) {
                return serverStreamingCall(request);
            }
            @Override
            public void close() throws Exception {
                channel.shutdown().awaitTermination(DEFAULT_TIMEOUT_SECONDS, SECONDS);
            }
            @Override
            public Completable closeAsync() {
                throw new UnsupportedOperationException();
            }
            @Override
            public Completable onClose() {
                throw new UnsupportedOperationException();
            }
            @Override
            public BlockingCompatClient asBlockingClient() {
                throw new UnsupportedOperationException();
            }
            private void sendRequest(final Publisher<CompatRequest> request,
                                     final StreamObserver<CompatRequest> requestObserver) {
                request.whenOnComplete(requestObserver::onCompleted)
                        .whenOnError(requestObserver::onError)
                        .forEach(requestObserver::onNext);
            }
            private StreamObserver<CompatResponse> adaptResponse(
                    final Processor<CompatResponse, CompatResponse> processor) {
                return new StreamObserver<CompatResponse>() {
                    @Override
                    public void onNext(final CompatResponse value) {
                        processor.onSuccess(value);
                    }
                    @Override
                    public void onError(final Throwable t) {
                        processor.onError(t);
                    }
                    @Override
                    public void onCompleted() {
                    }
                };
            }
            private StreamObserver<CompatResponse> adaptResponse(
                    final PublisherSource.Processor<CompatResponse, CompatResponse> processor) {
                return new StreamObserver<CompatResponse>() {
                    @Override
                    public void onNext(final CompatResponse value) {
                        processor.onNext(value);
                    }
                    @Override
                    public void onError(final Throwable t) {
                        processor.onError(t);
                    }
                    @Override
                    public void onCompleted() {
                        processor.onComplete();
                    }
                };
            }
        };
    }
private static CompatClient grpcJavaClient ( final SocketAddress address , @ Nullable final String compression , final boolean ssl , @ Nullable Duration timeout ) throws Exception { final NettyChannelBuilder builder = NettyChannelBuilder . forAddress ( address ) ; if ( ssl ) { final SslContext context = GrpcSslContexts . forClient ( ) . trustManager ( InsecureTrustManagerFactory . INSTANCE ) . build ( ) ; builder . sslContext ( context ) ; } else { builder . usePlaintext ( ) ; } final ManagedChannel channel = builder . build ( ) ; CompatGrpc . CompatStub stub = CompatGrpc . newStub ( channel ) ; if ( compression != null ) { stub = stub . withCompression ( compression ) ; } if ( null != timeout ) { stub = stub . withDeadlineAfter ( timeout . toNanos ( ) , NANOSECONDS ) ; } final CompatGrpc . CompatStub finalStub = stub ; return new CompatClient ( ) { @ Override public GrpcExecutionContext executionContext ( ) { throw new UnsupportedOperationException ( ) ; } @ Override public Publisher < CompatResponse > bidirectionalStreamingCall ( final Publisher < CompatRequest > request ) { final PublisherSource . Processor < CompatResponse , CompatResponse > processor = newPublisherProcessor ( 3 ) ; sendRequest ( request , finalStub . bidirectionalStreamingCall ( adaptResponse ( processor ) ) ) ; return fromSource ( processor ) ; } @ Override public Publisher < CompatResponse > bidirectionalStreamingCall ( final GrpcClientMetadata metadata , final Publisher < CompatRequest > request ) { return bidirectionalStreamingCall ( request ) ; } @ Deprecated @ Override public Publisher < CompatResponse > bidirectionalStreamingCall ( final BidirectionalStreamingCallMetadata metadata , final Publisher < CompatRequest > request ) { return bidirectionalStreamingCall ( request ) ; } @ SuppressWarnings ( "unchecked" ) @ Override public Single < CompatResponse > clientStreamingCall ( final Publisher < CompatRequest > request ) { final Processor < CompatResponse , CompatResponse > processor = newSingleProcessor ( ) ; final StreamObserver < CompatRequest > requestObserver = finalStub . clientStreamingCall ( adaptResponse ( processor ) ) ; sendRequest ( request , requestObserver ) ; return ( Single < CompatResponse > ) processor ; } @ Deprecated @ Override public Single < CompatResponse > clientStreamingCall ( final ClientStreamingCallMetadata metadata , final Publisher < CompatRequest > request ) { return clientStreamingCall ( request ) ; } @ Override public Single < CompatResponse > clientStreamingCall ( final GrpcClientMetadata metadata , final Publisher < CompatRequest > request ) { return clientStreamingCall ( request ) ; } @ SuppressWarnings ( "unchecked" ) @ Override public Single < CompatResponse > scalarCall ( final CompatRequest request ) { final Processor < CompatResponse , CompatResponse > processor = newSingleProcessor ( ) ; finalStub . scalarCall ( request , adaptResponse ( processor ) ) ; return ( Single < CompatResponse > ) processor ; } @ Deprecated @ Override public Single < CompatResponse > scalarCall ( final ScalarCallMetadata metadata , final CompatRequest request ) { return scalarCall ( request ) ; } @ Override public Single < CompatResponse > scalarCall ( final GrpcClientMetadata metadata , final CompatRequest request ) { return scalarCall ( request ) ; } @ Override public Publisher < CompatResponse > serverStreamingCall ( final CompatRequest request ) { final PublisherSource . Processor < CompatResponse , CompatResponse > processor = newPublisherProcessor ( 3 ) ; finalStub . serverStreamingCall ( request , adaptResponse ( processor ) ) ; return fromSource ( processor ) ; } @ Deprecated @ Override public Publisher < CompatResponse > serverStreamingCall ( final ServerStreamingCallMetadata metadata , final CompatRequest request ) { return serverStreamingCall ( request ) ; } @ Override public Publisher < CompatResponse > serverStreamingCall ( final GrpcClientMetadata metadata , final CompatRequest request ) { return serverStreamingCall ( request ) ; } @ Override public void close ( ) throws Exception { channel . shutdown ( ) . awaitTermination ( DEFAULT_TIMEOUT_SECONDS , SECONDS ) ; } @ Override public Completable closeAsync ( ) { throw new UnsupportedOperationException ( ) ; } @ Override public Completable onClose ( ) { throw new UnsupportedOperationException ( ) ; } @ Override public BlockingCompatClient asBlockingClient ( ) { throw new UnsupportedOperationException ( ) ; } private void sendRequest ( final Publisher < CompatRequest > request , final StreamObserver < CompatRequest > requestObserver ) { request . whenOnComplete ( requestObserver :: onCompleted ) . whenOnError ( requestObserver :: onError ) . forEach ( requestObserver :: onNext ) ; } private StreamObserver < CompatResponse > adaptResponse ( final Processor < CompatResponse , CompatResponse > processor ) { return new StreamObserver < CompatResponse > ( ) { @ Override public void onNext ( final CompatResponse value ) { processor . onSuccess ( value ) ; } @ Override public void onError ( final Throwable t ) { processor . onError ( t ) ; } @ Override public void onCompleted ( ) { } } ; } private StreamObserver < CompatResponse > adaptResponse ( final PublisherSource . Processor < CompatResponse , CompatResponse > processor ) { return new StreamObserver < CompatResponse > ( ) { @ Override public void onNext ( final CompatResponse value ) { processor . onNext ( value ) ; } @ Override public void onError ( final Throwable t ) { processor . onError ( t ) ; } @ Override public void onCompleted ( ) { processor . onComplete ( ) ; } } ; } } ; }
Wrap grpc client in our client interface to simplify test code
Wrap grpc client in our client interface to simplify test code
// stub is immutable and each builder step returns a new instance. // ignored