OperatorConditionalBinding implements Observable.Operator<T, T> { @Override public Subscriber<? super T> call(final Subscriber<? super T> child) { return new Subscriber<T>(child) { @Override public void onCompleted() { Assertions.assertUiThread(); if (shouldForwardNotification()) { child.onCompleted(); } else { handleLostBinding("onCompleted"); } } @Override public void onError(Throwable e) { Assertions.assertUiThread(); if (shouldForwardNotification()) { child.onError(e); } else { handleLostBinding("onError"); } } @Override public void onNext(T t) { Assertions.assertUiThread(); if (shouldForwardNotification()) { child.onNext(t); } else { handleLostBinding("onNext"); } } private boolean shouldForwardNotification() { return boundRef != null && predicate.call(boundRef); } private void handleLostBinding(String context) { log("bound object has become invalid; skipping " + context); log("unsubscribing..."); boundRef = null; unsubscribe(); } private void log(String message) { if (Log.isLoggable(LOG_TAG, Log.DEBUG)) { Log.d(LOG_TAG, message); } } }; } OperatorConditionalBinding(R bound, Func1<? super R, Boolean> predicate);  OperatorConditionalBinding(R bound); @Override Subscriber<? super T> call(final Subscriber<? super T> child);  }
@Test public void shouldUnsubscribeFromSourceSequenceWhenPredicateFailsToPass() { OperatorConditionalBinding<String, Object> op = new OperatorConditionalBinding<String, Object>( new Object(), UtilityFunctions.alwaysFalse()); Subscriber<? super String> sub = op.call(subscriber); sub.onNext("one"); sub.onNext("two"); sub.onCompleted(); sub.onError(new Exception()); assertEquals(0, subscriber.getOnNextEvents().size()); assertEquals(0, subscriber.getOnCompletedEvents().size()); assertEquals(0, subscriber.getOnErrorEvents().size()); }
@Test public void unsubscribeWillUnsubscribeFromWrappedSubscriber() { OperatorConditionalBinding<String, Object> op = new OperatorConditionalBinding<String, Object>(new Object()); op.call(subscriber).unsubscribe(); subscriber.assertUnsubscribed(); }