-
Notifications
You must be signed in to change notification settings - Fork 7.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #938 from soundcloud/operator-weak-binding
OperatorWeakBinding (deprecates OperatorObserveFromAndroidComponent)
- Loading branch information
Showing
5 changed files
with
283 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorWeakBinding.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package rx.operators; | ||
|
|
||
| import rx.Observable; | ||
| import rx.Subscriber; | ||
| import rx.functions.Func1; | ||
| import rx.functions.Functions; | ||
|
|
||
| import android.util.Log; | ||
|
|
||
| import java.lang.ref.WeakReference; | ||
|
|
||
| /** | ||
| * Ties a source sequence to the life-cycle of the given target object, and/or the subscriber | ||
| * using weak references. When either object is gone, this operator automatically unsubscribes | ||
| * from the source sequence. | ||
| * <p/> | ||
| * You can also pass in an optional predicate function, which whenever it evaluates to false | ||
| * on the target object, will also result in the operator unsubscribing from the sequence. | ||
| * | ||
| * @param <T> the type of the objects emitted to a subscriber | ||
| * @param <R> the type of the target object to bind to | ||
| */ | ||
| public final class OperatorWeakBinding<T, R> implements Observable.Operator<T, T> { | ||
|
|
||
| private static final String LOG_TAG = "WeakBinding"; | ||
|
|
||
| final WeakReference<R> boundRef; | ||
| private final Func1<? super R, Boolean> predicate; | ||
|
|
||
| public OperatorWeakBinding(R bound, Func1<? super R, Boolean> predicate) { | ||
| boundRef = new WeakReference<R>(bound); | ||
| this.predicate = predicate; | ||
| } | ||
|
|
||
| public OperatorWeakBinding(R bound) { | ||
| boundRef = new WeakReference<R>(bound); | ||
| this.predicate = Functions.alwaysTrue(); | ||
| } | ||
|
|
||
| @Override | ||
| public Subscriber<? super T> call(final Subscriber<? super T> child) { | ||
| return new WeakSubscriber(child); | ||
| } | ||
|
|
||
| final class WeakSubscriber extends Subscriber<T> { | ||
|
|
||
| final WeakReference<Subscriber<? super T>> subscriberRef; | ||
|
|
||
| private WeakSubscriber(Subscriber<? super T> source) { | ||
| super(source); | ||
| subscriberRef = new WeakReference<Subscriber<? super T>>(source); | ||
| } | ||
|
|
||
| @Override | ||
| public void onCompleted() { | ||
| final Subscriber<? super T> sub = subscriberRef.get(); | ||
| if (shouldForwardNotification(sub)) { | ||
| sub.onCompleted(); | ||
| } else { | ||
| handleLostBinding(sub, "onCompleted"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable e) { | ||
| final Subscriber<? super T> sub = subscriberRef.get(); | ||
| if (shouldForwardNotification(sub)) { | ||
| sub.onError(e); | ||
| } else { | ||
| handleLostBinding(sub, "onError"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| final Subscriber<? super T> sub = subscriberRef.get(); | ||
| if (shouldForwardNotification(sub)) { | ||
| sub.onNext(t); | ||
| } else { | ||
| handleLostBinding(sub, "onNext"); | ||
| } | ||
| } | ||
|
|
||
| private boolean shouldForwardNotification(Subscriber<? super T> sub) { | ||
| final R target = boundRef.get(); | ||
| return sub != null && target != null && predicate.call(target); | ||
| } | ||
|
|
||
| private void handleLostBinding(Subscriber<? super T> sub, String context) { | ||
| if (sub == null) { | ||
| log("subscriber gone; skipping " + context); | ||
| } else { | ||
| final R r = boundRef.get(); | ||
| if (r != null) { | ||
| // the predicate failed to validate | ||
| log("bound component has become invalid; skipping " + context); | ||
| } else { | ||
| log("bound component gone; skipping " + context); | ||
| } | ||
| } | ||
| log("unsubscribing..."); | ||
| unsubscribe(); | ||
| } | ||
|
|
||
| private void log(String message) { | ||
| if (Log.isLoggable(LOG_TAG, Log.DEBUG)) { | ||
| Log.d(LOG_TAG, message); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
94 changes: 94 additions & 0 deletions
94
rxjava-contrib/rxjava-android/src/test/java/rx/operators/OperatorWeakBindingTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package rx.operators; | ||
|
|
||
| import static org.junit.Assert.assertEquals; | ||
|
|
||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
| import org.junit.runner.RunWith; | ||
| import org.mockito.MockitoAnnotations; | ||
| import org.robolectric.RobolectricTestRunner; | ||
| import rx.functions.Functions; | ||
| import rx.observers.TestSubscriber; | ||
|
|
||
| import java.util.Arrays; | ||
|
|
||
| @RunWith(RobolectricTestRunner.class) | ||
| public class OperatorWeakBindingTest { | ||
|
|
||
| private TestSubscriber<String> subscriber = new TestSubscriber<String>(); | ||
|
|
||
| @Before | ||
| public void setUp() throws Exception { | ||
| MockitoAnnotations.initMocks(this); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldForwardAllNotificationsWhenSubscriberAndTargetAlive() { | ||
| OperatorWeakBinding<String, Object> op = new OperatorWeakBinding<String, Object>(new Object()); | ||
| OperatorWeakBinding.WeakSubscriber weakSub = (OperatorWeakBinding.WeakSubscriber) op.call(subscriber); | ||
| weakSub.onNext("one"); | ||
| weakSub.onNext("two"); | ||
| weakSub.onCompleted(); | ||
| weakSub.onError(new Exception()); | ||
|
|
||
| subscriber.assertReceivedOnNext(Arrays.asList("one", "two")); | ||
| assertEquals(1, subscriber.getOnCompletedEvents().size()); | ||
| assertEquals(1, subscriber.getOnErrorEvents().size()); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldUnsubscribeFromSourceSequenceWhenSubscriberReleased() { | ||
| OperatorWeakBinding<String, Object> op = new OperatorWeakBinding<String, Object>(new Object()); | ||
|
|
||
| OperatorWeakBinding.WeakSubscriber weakSub = (OperatorWeakBinding.WeakSubscriber) op.call(subscriber); | ||
| weakSub.onNext("one"); | ||
| weakSub.subscriberRef.clear(); | ||
| weakSub.onNext("two"); | ||
| weakSub.onCompleted(); | ||
| weakSub.onError(new Exception()); | ||
|
|
||
| subscriber.assertReceivedOnNext(Arrays.asList("one")); | ||
| assertEquals(0, subscriber.getOnCompletedEvents().size()); | ||
| assertEquals(0, subscriber.getOnErrorEvents().size()); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldUnsubscribeFromSourceSequenceWhenTargetObjectReleased() { | ||
| OperatorWeakBinding<String, Object> op = new OperatorWeakBinding<String, Object>(new Object()); | ||
|
|
||
| OperatorWeakBinding.WeakSubscriber weakSub = (OperatorWeakBinding.WeakSubscriber) op.call(subscriber); | ||
| weakSub.onNext("one"); | ||
| op.boundRef.clear(); | ||
| weakSub.onNext("two"); | ||
| weakSub.onCompleted(); | ||
| weakSub.onError(new Exception()); | ||
|
|
||
| subscriber.assertReceivedOnNext(Arrays.asList("one")); | ||
| assertEquals(0, subscriber.getOnCompletedEvents().size()); | ||
| assertEquals(0, subscriber.getOnErrorEvents().size()); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldUnsubscribeFromSourceSequenceWhenPredicateFailsToPass() { | ||
| OperatorWeakBinding<String, Object> op = new OperatorWeakBinding<String, Object>( | ||
| new Object(), Functions.alwaysFalse()); | ||
|
|
||
| OperatorWeakBinding.WeakSubscriber weakSub = (OperatorWeakBinding.WeakSubscriber) op.call(subscriber); | ||
| weakSub.onNext("one"); | ||
| weakSub.onNext("two"); | ||
| weakSub.onCompleted(); | ||
| weakSub.onError(new Exception()); | ||
|
|
||
| assertEquals(0, subscriber.getOnNextEvents().size()); | ||
| assertEquals(0, subscriber.getOnCompletedEvents().size()); | ||
| assertEquals(0, subscriber.getOnErrorEvents().size()); | ||
| } | ||
|
|
||
| @Test | ||
| public void unsubscribeWillUnsubscribeFromWrappedSubscriber() { | ||
| OperatorWeakBinding<String, Object> op = new OperatorWeakBinding<String, Object>(new Object()); | ||
|
|
||
| op.call(subscriber).unsubscribe(); | ||
| subscriber.assertUnsubscribed(); | ||
| } | ||
| } |