New RIPC abstractions

TcpConnection.java

public interface TcpConnection<R, W> {

  Reader<R> reader();

  Writer<W> writer();

  public interface Reader<R> extends Publisher<R> {
    <NEWR> Reader<NEWR> intercept(Interceptor<R, NEWR> interceptor);
  }

  public interface Writer<W> extends Processor<W, W> {
    <NEWW> Writer<NEWW> intercept(Interceptor<W, NEWW> interceptor);

    void flush();
  }

}

TcpConnectionHandler.java

public interface TcpConnectionHandler<R, W> {

  void onOpen(TcpConnection<R, W> connection);

  void onAbort(TcpConnection<R, W> connection);

  void onClose(TcpConnection<R, W> connection);

  void onError(TcpConnection<R, W> connection);

  void onReadable(TcpConnection<R, W> connection);

  void onWritable(TcpConnection<R, W> connection);

}

TcpConnectionHandlerOnOpen.java

@Override
public void onOpen(TcpConnection<ByteBuf, ByteBuf> connection) {
  connection
      .reader()
      .intercept(new Interceptor<ByteBuf, String>() {
        @Override
        public String intercept(ByteBuf obj) {
          return obj.toString(Charset.defaultCharset());
        }
      })
      .subscribe(new Subscriber<String>() {
        @Override
        public void onSubscribe(Subscription s) {

        }

        @Override
        public void onNext(String s) {

        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {

        }
      });
}