@Test public void pipelinedReadsCompleteBeforeWrites() throws InterruptedException { toSource(requester.write(writePublisher1)).subscribe(readSubscriber); readSubscriber.awaitSubscription().request(1); toSource(requester.write(writePublisher2)).subscribe(readSubscriber2); channel.writeInbound(1); channel.writeInbound(2); Integer next = readSubscriber.takeOnNext(); assertNotNull(next); assertEquals(next.intValue(), 1); assertTrue(writePublisher1.isSubscribed()); assertFalse(writePublisher2.isSubscribed()); writePublisher1.onNext(1); writePublisher1.onComplete(); readSubscriber.awaitOnComplete(); readSubscriber2.awaitSubscription().request(1); next = readSubscriber2.takeOnNext(); assertNotNull(next); assertEquals(next.intValue(), 2); Integer channelWrite = channel.readOutbound(); assertNotNull(channelWrite); assertEquals(channelWrite.intValue(), 1); assertTrue(writePublisher2.isSubscribed()); writePublisher2.onNext(2); writePublisher2.onComplete(); readSubscriber2.awaitOnComplete(); channelWrite = channel.readOutbound(); assertNotNull(channelWrite); assertEquals(channelWrite.intValue(), 2); }
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();  }