ClosedChannelException while TcpClient reads from TcpServer about rxnetty HOT 3 OPEN

import io.netty.buffer.ByteBuf;
import io.reactivex.netty.channel.Connection;
import io.reactivex.netty.protocol.tcp.client.TcpClient;
import io.reactivex.netty.protocol.tcp.server.TcpServer;
import org.testng.annotations.Test;
import rx.Observable;
import rx.observables.StringObservable;
import rx.observers.TestSubscriber;

import java.nio.channels.ClosedChannelException;
import java.nio.charset.Charset;
import java.util.stream.Stream;

public class TcpTest {
  @Test
  public void testSinkClosedChannelException() throws Exception {
    int count = 100;
    TcpServer<ByteBuf, ByteBuf> server = TcpServer.newServer(0)
      .start(
        c ->
          c.writeString(
            Observable.range(0, count)
              .map(i -> i + ",")
          )
      );

    TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
    TcpClient.newClient("127.0.0.1", server.getServerPort())
      .createConnectionRequest()
      .flatMap(Connection::getInput)
      .map(
        b -> {
          // must free the ByteBuf
          String s = b.toString(Charset.defaultCharset());
          b.release();
          return s;
        }
      )
      // can't guarantee that the block boundaries will line up so we force split
      .compose(o -> StringObservable.split(o, ","))
      .map(Integer::parseInt)
      .onErrorResumeNext(
        e -> {
          if (e instanceof ClosedChannelException) {
            // sink ClosedChannelException to nothing
            return Observable.empty();
          }
          // 'rethrow' all other errors
          return Observable.error(e);
        }
      )
      .subscribe(testSubscriber);

    testSubscriber.awaitTerminalEvent();

    testSubscriber.assertValueCount(count);
    testSubscriber.assertValues(
      Stream.iterate(0, i -> i + 1).limit(count).toArray(Integer[]::new)
    );
    testSubscriber.assertNoErrors();
}