@Test public void flushStrategy() throws InterruptedException { FlushStrategy flushStrategy1 = mock(FlushStrategy.class); FlushStrategy.WriteEventsListener writeEventsListener1 = mock(FlushStrategy.WriteEventsListener.class); AtomicReference<FlushStrategy.FlushSender> sender1Ref = new AtomicReference<>(); doAnswer((Answer<FlushStrategy.WriteEventsListener>) invocation -> { sender1Ref.compareAndSet(null, invocation.getArgument(0, FlushStrategy.FlushSender.class)); return writeEventsListener1; }).when(flushStrategy1).apply(any()); FlushStrategy flushStrategy2 = mock(FlushStrategy.class); FlushStrategy.WriteEventsListener writeEventsListener2 = mock(FlushStrategy.WriteEventsListener.class); AtomicReference<FlushStrategy.FlushSender> sender2Ref = new AtomicReference<>(); doAnswer((Answer<FlushStrategy.WriteEventsListener>) invocation -> { sender2Ref.compareAndSet(null, invocation.getArgument(0, FlushStrategy.FlushSender.class)); return writeEventsListener2; }).when(flushStrategy2).apply(any()); toSource(requester.write(writePublisher1, () -> flushStrategy1, WriteDemandEstimators::newDefaultEstimator)) .subscribe(readSubscriber); toSource(requester.write(writePublisher2, () -> flushStrategy2, WriteDemandEstimators::newDefaultEstimator)) .subscribe(readSubscriber2); readSubscriber.awaitSubscription().request(1); assertTrue(writePublisher1.isSubscribed()); assertFalse(writePublisher2.isSubscribed()); verify(writeEventsListener1).writeStarted(); FlushStrategy.FlushSender sender1 = sender1Ref.get(); assertNotNull(sender1); writePublisher1.onNext(1); sender1.flush(); verify(writeEventsListener1).itemWritten(eq(1)); writePublisher1.onComplete(); verify(writeEventsListener1).writeTerminated(); assertTrue(writePublisher2.isSubscribed()); verify(writeEventsListener2).writeStarted(); FlushStrategy.FlushSender sender2 = sender1Ref.get(); assertNotNull(sender2); writePublisher2.onNext(2); sender2.flush(); verify(writeEventsListener2).itemWritten(eq(2)); writePublisher2.onComplete(); verify(writeEventsListener2).writeTerminated(); }
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();  }