@Test public void writeSubscribeThrowsLetsSubsequentRequestsThrough() throws InterruptedException { AtomicBoolean firstReadOperation = new AtomicBoolean(); TestPublisher<Integer> mockReadPublisher1 = new TestPublisher<>(); TestPublisher<Integer> mockReadPublisher2 = new TestPublisher<>(); @SuppressWarnings("unchecked") NettyConnection<Integer, Integer> mockConnection = mock(NettyConnection.class); doAnswer((Answer<Publisher<Integer>>) invocation -> firstReadOperation.compareAndSet(false, true) ? mockReadPublisher1 : mockReadPublisher2 ).when(mockConnection).read(); doAnswer((Answer<Completable>) invocation -> new Completable() { @Override protected void handleSubscribe(final CompletableSource.Subscriber subscriber) { throw DELIBERATE_EXCEPTION; } }).when(mockConnection).write(eq(writePublisher1), any(), any()); doAnswer((Answer<Completable>) invocation -> { Publisher<Integer> writePub = invocation.getArgument(0); return writePub.ignoreElements(); }).when(mockConnection).write(eq(writePublisher2), any(), any()); requester = new NettyPipelinedConnection<>(mockConnection); toSource(requester.write(writePublisher1)).subscribe(readSubscriber); toSource(requester.write(writePublisher2)).subscribe(readSubscriber2); Subscription readSubscription = readSubscriber.awaitSubscription(); readSubscription.request(1); assertTrue(mockReadPublisher1.isSubscribed()); assertThat(readSubscriber.awaitOnError(), is(DELIBERATE_EXCEPTION)); assertFalse(writePublisher1.isSubscribed()); verifySecondRequestProcessed(mockReadPublisher2, mockConnection); }
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();  }