@Test public void delegatesDisposeToActualConnection() throws Exception { connection = underTest.connect(recordingConsumer); connection.dispose(); assertThat(testConnection.disposed, is(true)); }
@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 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 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); }
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); @Nonnull @Override Connection<I> connect(Consumer<O> output); }
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); @Nonnull @Override Connection<I> connect(Consumer<O> output);  }