@Test public void testChannelCloseAfterWriteComplete() { writeListener.listen(conn.write(publisher)); Buffer hello1 = newBuffer("Hello1"); publisher.onNext(hello1); publisher.onNext(TRAILER.duplicate()); pollChannelAndVerifyWrites("Hello1", TRAILER_MSG); channel.pipeline().fireChannelInactive(); publisher.onComplete(); channel.close(); writeListener.verifyCompletion(); }
@Override public Completable write(Publisher<Write> write) { return write(write, flushStrategyHolder::currentStrategy, WriteDemandEstimators::newDefaultEstimator); }
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Completable write(Publisher<Write> write) { return write(write, flushStrategyHolder::currentStrategy, WriteDemandEstimators::newDefaultEstimator); } }
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Completable write(Publisher<Write> write) { return write(write, flushStrategyHolder::currentStrategy, WriteDemandEstimators::newDefaultEstimator); } private  DefaultNettyConnection(Channel channel, BufferAllocator allocator, Executor executor,
                                   Predicate<Read> terminalPredicate, CloseHandler closeHandler,
                                   FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
                                   ExecutionStrategy executionStrategy, Protocol protocol,
                                   @Nullable SSLSession sslSession, @Nullable ChannelConfig parentChannelConfig,
                                   DataObserver dataObserver); }
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Completable write(Publisher<Write> write) { return write(write, flushStrategyHolder::currentStrategy, WriteDemandEstimators::newDefaultEstimator); } private  DefaultNettyConnection(Channel channel, BufferAllocator allocator, Executor executor,
                                   Predicate<Read> terminalPredicate, CloseHandler closeHandler,
                                   FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
                                   ExecutionStrategy executionStrategy, Protocol protocol,
                                   @Nullable SSLSession sslSession, @Nullable ChannelConfig parentChannelConfig,
                                   DataObserver dataObserver); static DefaultNettyConnection<Read, Write> initChildChannel(
            Channel channel, BufferAllocator allocator, Executor executor, Predicate<Read> terminalPredicate,
            CloseHandler closeHandler, FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
            ExecutionStrategy executionStrategy, Protocol protocol, @Nullable SSLSession sslSession,
            @Nullable ChannelConfig parentChannelConfig, StreamObserver streamObserver); static Single<DefaultNettyConnection<Read, Write>> initChannel(
            Channel channel, BufferAllocator allocator, Executor executor, Predicate<Read> terminalPredicate,
            CloseHandler closeHandler, FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
            ChannelInitializer initializer, ExecutionStrategy executionStrategy, Protocol protocol,
            ConnectionObserver observer); @Override Publisher<Read> read(); @Override Completable write(Publisher<Write> write); @Override Completable write(final Publisher<Write> write,
                             final Supplier<FlushStrategy> flushStrategySupplier,
                             final Supplier<WriteDemandEstimator> demandEstimatorSupplier); @Override SocketAddress localAddress(); @Override SocketAddress remoteAddress(); @Override SSLSession sslSession(); @Override ExecutionContext executionContext(); @Nullable @Override T socketOption(final SocketOption<T> option); @Override Protocol protocol(); @Override Completable onClosing(); @Override Channel nettyChannel(); @Override String toString(); @Override Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider); @Override FlushStrategy defaultFlushStrategy(); @Override Single<Throwable> transportError(); }
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Completable write(Publisher<Write> write) { return write(write, flushStrategyHolder::currentStrategy, WriteDemandEstimators::newDefaultEstimator); } private  DefaultNettyConnection(Channel channel, BufferAllocator allocator, Executor executor,
                                   Predicate<Read> terminalPredicate, CloseHandler closeHandler,
                                   FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
                                   ExecutionStrategy executionStrategy, Protocol protocol,
                                   @Nullable SSLSession sslSession, @Nullable ChannelConfig parentChannelConfig,
                                   DataObserver dataObserver); static DefaultNettyConnection<Read, Write> initChildChannel(
            Channel channel, BufferAllocator allocator, Executor executor, Predicate<Read> terminalPredicate,
            CloseHandler closeHandler, FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
            ExecutionStrategy executionStrategy, Protocol protocol, @Nullable SSLSession sslSession,
            @Nullable ChannelConfig parentChannelConfig, StreamObserver streamObserver); static Single<DefaultNettyConnection<Read, Write>> initChannel(
            Channel channel, BufferAllocator allocator, Executor executor, Predicate<Read> terminalPredicate,
            CloseHandler closeHandler, FlushStrategy flushStrategy, @Nullable Long idleTimeoutMs,
            ChannelInitializer initializer, ExecutionStrategy executionStrategy, Protocol protocol,
            ConnectionObserver observer); @Override Publisher<Read> read(); @Override Completable write(Publisher<Write> write); @Override Completable write(final Publisher<Write> write,
                             final Supplier<FlushStrategy> flushStrategySupplier,
                             final Supplier<WriteDemandEstimator> demandEstimatorSupplier); @Override SocketAddress localAddress(); @Override SocketAddress remoteAddress(); @Override SSLSession sslSession(); @Override ExecutionContext executionContext(); @Nullable @Override T socketOption(final SocketOption<T> option); @Override Protocol protocol(); @Override Completable onClosing(); @Override Channel nettyChannel(); @Override String toString(); @Override Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider); @Override FlushStrategy defaultFlushStrategy(); @Override Single<Throwable> transportError();  }