@Test public void readCancelClosesConnectionThenWriteDoesNotSubscribe() throws InterruptedException { TestSubscription writePublisher1Subscription = new TestSubscription(); toSource(requester.write(writePublisher1 .afterSubscription(() -> writePublisher1Subscription))).subscribe(readSubscriber); Subscription readSubscription = readSubscriber.awaitSubscription(); readSubscription.request(1); assertTrue(writePublisher1.isSubscribed()); readSubscription.cancel(); writePublisher1Subscription.awaitCancelledUninterruptibly(); assertFalse(channel.isOpen()); toSource(requester.write(writePublisher2)).subscribe(readSubscriber2); assertThat(readSubscriber2.awaitOnError(), is(instanceOf(ClosedChannelException.class))); assertFalse(writePublisher2.isSubscribed()); }
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();  }