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);  }
@Test public void nullConsumerInConnectThrowsNPE() throws Exception { assertThatThrownBy(() -> underTest.connect(null)).isInstanceOf(NullPointerException.class); }
@Test public void nullDisposableConsumerReturnedToConnectThrowsNPE() throws Exception { underTest = new DiscardAfterDisposeConnectable<>( new Connectable<Integer, String>() { @Nonnull @Override public Connection<Integer> connect(Consumer<String> output) { return null; } }); assertThatThrownBy(() -> underTest.connect(recordingConsumer)) .isInstanceOf(NullPointerException.class); }
@Test public void forwardsMessagesToWrappedConsumer() throws Exception { connection = underTest.connect(recordingConsumer); connection.accept(14); recordingConsumer.assertValues("Value is: 14"); }
@Test public void delegatesDisposeToActualConnection() throws Exception { connection = underTest.connect(recordingConsumer); connection.dispose(); assertThat(testConnection.disposed, is(true)); }
@Test public void discardsEventsAfterDisposal() throws Exception { connection = underTest.connect(recordingConsumer); connection.dispose(); connection.accept(1); recordingConsumer.assertValues(); }