@Test public void pipelinedWriteAndReadCompleteSequential() throws InterruptedException { toSource(requester.write(writePublisher1)).subscribe(readSubscriber); readSubscriber.awaitSubscription().request(1); toSource(requester.write(writePublisher2)).subscribe(readSubscriber2); assertTrue(writePublisher1.isSubscribed()); assertFalse(writePublisher2.isSubscribed()); writePublisher1.onNext(1); writePublisher1.onComplete(); channel.writeInbound(1); Integer next = readSubscriber.takeOnNext(); assertNotNull(next); assertEquals(next.intValue(), 1); readSubscriber.awaitOnComplete(); readSubscriber2.awaitSubscription().request(1); writePublisher2.onNext(1); writePublisher2.onComplete(); channel.writeInbound(2); next = readSubscriber2.takeOnNext(); assertNotNull(next); assertEquals(next.intValue(), 2); readSubscriber2.awaitOnComplete(); }
Publisher<Resp> write(final Publisher<Req> requestPublisher) { return write(requestPublisher, connection::defaultFlushStrategy, WriteDemandEstimators::newDefaultEstimator); }
NettyPipelinedConnection implements NettyConnectionContext { Publisher<Resp> write(final Publisher<Req> requestPublisher) { return write(requestPublisher, connection::defaultFlushStrategy, WriteDemandEstimators::newDefaultEstimator); } }
NettyPipelinedConnection implements NettyConnectionContext { Publisher<Resp> write(final Publisher<Req> requestPublisher) { return write(requestPublisher, connection::defaultFlushStrategy, WriteDemandEstimators::newDefaultEstimator); } NettyPipelinedConnection(NettyConnection<Resp, Req> connection); }
NettyPipelinedConnection implements NettyConnectionContext { Publisher<Resp> write(final Publisher<Req> requestPublisher) { return write(requestPublisher, connection::defaultFlushStrategy, WriteDemandEstimators::newDefaultEstimator); } NettyPipelinedConnection(NettyConnection<Resp, Req> connection); @Override SocketAddress localAddress(); @Override SocketAddress remoteAddress(); @Override @Nullable SSLSession sslSession(); @Override ExecutionContext executionContext(); @Nullable @Override T socketOption(final SocketOption<T> option); @Override Protocol protocol(); @Override Single<Throwable> transportError(); @Override Completable onClosing(); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); @Override Channel nettyChannel(); @Override String toString(); @Override Cancellable updateFlushStrategy(final NettyConnectionContext.FlushStrategyProvider strategyProvider); @Override FlushStrategy defaultFlushStrategy(); }
NettyPipelinedConnection implements NettyConnectionContext { Publisher<Resp> write(final Publisher<Req> requestPublisher) { return write(requestPublisher, connection::defaultFlushStrategy, WriteDemandEstimators::newDefaultEstimator); } NettyPipelinedConnection(NettyConnection<Resp, Req> connection); @Override SocketAddress localAddress(); @Override SocketAddress remoteAddress(); @Override @Nullable SSLSession sslSession(); @Override ExecutionContext executionContext(); @Nullable @Override T socketOption(final SocketOption<T> option); @Override Protocol protocol(); @Override Single<Throwable> transportError(); @Override Completable onClosing(); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); @Override Channel nettyChannel(); @Override String toString(); @Override Cancellable updateFlushStrategy(final NettyConnectionContext.FlushStrategyProvider strategyProvider); @Override FlushStrategy defaultFlushStrategy();  }