### Question:
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider) { return flushStrategyHolder.updateFlushStrategy(strategyProvider); } 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();  }### Answer:
@Test public void testUpdateFlushStrategy() { writeListener.listen(conn.write(from(newBuffer("Hello"), TRAILER.duplicate()))); writeListener.verifyCompletion(); pollChannelAndVerifyWrites("Hello", TRAILER_MSG); writeListener.reset(); Cancellable c = conn.updateFlushStrategy((old, __) -> batchFlush(3, never())); writeListener.listen(conn.write(publisher)); publisher.onNext(newBuffer("Hello1")); pollChannelAndVerifyWrites(); publisher.onNext(newBuffer("Hello2")); publisher.onNext(TRAILER.duplicate()); pollChannelAndVerifyWrites("Hello1", "Hello2", TRAILER_MSG); publisher.onComplete(); writeListener.verifyCompletion(); c.cancel(); writeListener.reset(); writeListener.listen(conn.write(from(newBuffer("Hello3"), TRAILER.duplicate()))); writeListener.verifyCompletion(); pollChannelAndVerifyWrites("Hello3", TRAILER_MSG); }