class DiscardAfterDisposeConnectable implements Connectable<I, O> {

  @Nonnull
  @Override
  public Connection<I> connect(Consumer<O> output) {
    final DiscardAfterDisposeWrapper<O> safeOutput = wrapConsumer(output);
    final Connection<I> input = actual.connect(safeOutput);
    final DiscardAfterDisposeWrapper<I> safeInput = wrapConnection(input);

    final Disposable disposable = CompositeDisposable.from(safeInput, safeOutput);

    return new Connection<I>() {
      @Override
      public void accept(I effect) {
        safeInput.accept(effect);
      }

      @Override
      public void dispose() {
        disposable.dispose();
      }
    };
  }

    DiscardAfterDisposeConnectable(Connectable<I, O> actual);

}