### Question:
DefaultNettyConnection extends NettyChannelListenableAsyncCloseable implements NettyConnection<Read, Write> { @Override public Publisher<Read> read() { return readPublisher; } 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 testExceptionWithNoSubscriberIsQueued() throws Exception { channel.pipeline().fireExceptionCaught(DELIBERATE_EXCEPTION); toSource(conn.read()).subscribe(subscriber); assertThat(subscriber.takeError(), is(DELIBERATE_EXCEPTION)); conn.onClose().toFuture().get(); }
@Test public void testChannelInactiveWithNoSubscriberIsQueued() throws Exception { channel.close().get(); toSource(conn.read()).subscribe(subscriber); assertThat(subscriber.takeError(), instanceOf(ClosedChannelException.class)); conn.onClose().toFuture().get(); }
@Test public void testRead() { toSource(conn.read()).subscribe(subscriber); Buffer expected = allocator.fromAscii("data"); channel.writeInbound(expected.duplicate()); assertThat(subscriber.takeItems(), hasSize(0)); assertThat(subscriber.takeTerminal(), nullValue()); subscriber.request(1); assertThat(subscriber.takeItems(), contains(expected)); assertThat(subscriber.takeTerminal(), is(complete())); }
@Test public void testTerminalPredicateThrowTerminatesReadPublisher() throws Exception { CloseHandler closeHandler = forPipelinedRequestResponse(true, channel.config()); setupWithCloseHandler(closeHandler); toSource(conn.read()).subscribe(subscriber); subscriber.request(1); channel.writeInbound(allocator.fromAscii("DELIBERATE_EXCEPTION")); closeHandler.channelClosedInbound(channel.pipeline().firstContext()); assertThat(subscriber.takeError(), is(DELIBERATE_EXCEPTION)); }