Add deliverLatestToView to rx2/RxTiPresenterUtils

thirtyinch-rx2/src/main/java/net/grandcentrix/thirtyinch/rx2/RxTiPresenterUtils.java

@@ -18,14 +18,34 @@
 import io.reactivex.Observable;
 import io.reactivex.ObservableEmitter;
 import io.reactivex.ObservableOnSubscribe;
+import io.reactivex.ObservableSource;
+import io.reactivex.ObservableTransformer;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.functions.BiFunction;
+import io.reactivex.functions.Function;
 import net.grandcentrix.thirtyinch.Removable;
 import net.grandcentrix.thirtyinch.TiLifecycleObserver;
 import net.grandcentrix.thirtyinch.TiPresenter;
 import net.grandcentrix.thirtyinch.TiView;
 
 public class RxTiPresenterUtils {
 
+    /**
+     * Wrapper for an emitted value, along with a record of whether or not the view was attached when the value was
+     * emitted.
+     */
+    private static class ViewReadyValue<T> {
+
+        T value;
+
+        boolean viewReady;
+
+        ViewReadyValue(final T t, final Boolean viewReady) {
+            this.value = t;
+            this.viewReady = viewReady;
+        }
+    }
+
     /**
      * Observable of the view state. The View is ready to receive calls after calling {@link
      * TiPresenter#attachView(TiView)} and before calling {@link TiPresenter#detachView()}.
@@ -66,4 +86,54 @@ public boolean isDisposed() {
         }).distinctUntilChanged();
     }
 
+    /**
+     * Returns a transformer that will delay onNext, onError and onComplete emissions unless a view
+     * become available. getView() is guaranteed to be != null during all emissions. This
+     * transformer can only be used on application's main thread.
+     * <p/>
+     * If this transformer receives a next value while the previous value has not been delivered,
+     * the previous value will be dropped.
+     * <p/>
+     * Use this operator when you need to show updatable data.
+     *
+     * @param <T>       a type of onNext value.
+     * @param presenter the presenter waiting for the view
+     * @return the delaying operator.
+     */
+    public static <T> ObservableTransformer<T, T> deliverLatestToView(
+            final TiPresenter presenter) {
+        return new ObservableTransformer<T, T>() {
+            @Override
+            public ObservableSource<T> apply(final Observable<T> observable) {
+
+                // make sure we never complete
+                final Observable<T> source = observable.concatWith(Observable.<T>never());
+
+                // The order of the sources is important here! We want the viewReady emission to be captured first so that any synchronous
+                // source emissions are not skipped.
+                // See https://github.com/ReactiveX/RxJava/issues/5325
+                return Observable
+                        .combineLatest(isViewReady(presenter), source,
+                                new BiFunction<Boolean, T, ViewReadyValue<T>>() {
+                                    @Override
+                                    public ViewReadyValue<T> apply(final Boolean viewReady, final T t)
+                                            throws Exception {
+                                        return new ViewReadyValue<>(t, viewReady);
+                                    }
+                                })
+                        .flatMap(new Function<ViewReadyValue<T>, ObservableSource<T>>() {
+                            @Override
+                            public ObservableSource<T> apply(final ViewReadyValue<T> viewReadyValue)
+                                    throws Exception {
+                                if (viewReadyValue.viewReady) {
+                                    return Observable.just(viewReadyValue.value);
+                                } else {
+                                    return Observable.empty();
+                                }
+                            }
+                        });
+            }
+        };
+    }
+
 }

thirtyinch-rx2/src/test/java/net/grandcentrix/thirtyinch/rx2/RxTiPresenterUtilsTest.java

@@ -2,7 +2,9 @@
 
 import static org.mockito.Mockito.*;
 
+import io.reactivex.Observable;
 import io.reactivex.observers.TestObserver;
+import io.reactivex.subjects.PublishSubject;
 import net.grandcentrix.thirtyinch.TiPresenter;
 import net.grandcentrix.thirtyinch.TiView;
 import org.junit.*;
@@ -57,4 +59,116 @@ public void testIsViewReady_DisposeBeforeAttachView_ShouldRemoveCallback() throw
         mPresenter.attachView(mView);
         test.assertValue(false);
     }
-}
+
+    @Test
+    public void testDeliverLatestToView_ViewNotReady() throws Exception {
+        mPresenter.create();
+
+        TestObserver<Integer> testObserver = new TestObserver<>();
+        Observable.just(1, 2, 3)
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        mPresenter.attachView(mView);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertValuesOnly(3);
+    }
+
+    @Test
+    public void testDeliverLatestToView_ViewReady() throws Exception {
+        mPresenter.create();
+        mPresenter.attachView(mView);
+
+        TestObserver<Integer> testObserver = new TestObserver<>();
+        Observable.just(1, 2, 3)
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertValuesOnly(1, 2, 3);
+    }
+
+    @Test
+    public void testDeliverLatestToView_ViewNeverReady() throws Exception {
+        mPresenter.create();
+
+        TestObserver<Integer> testObserver = new TestObserver<>();
+        Observable.just(1, 2, 3)
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertEmpty();
+    }
+
+    @Test
+    public void testDeliverLatestToView_ViewComesAndGoes() throws Exception {
+        mPresenter.create();
+
+        PublishSubject<Integer> source = PublishSubject.create();
+        TestObserver<Integer> testObserver = new TestObserver<>();
+
+        source
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        source.onNext(1);
+        source.onNext(2);
+        mPresenter.attachView(mView);
+        source.onNext(3);
+        mPresenter.detachView();
+        source.onNext(4);
+        source.onNext(5);
+        mPresenter.attachView(mView);
+        source.onNext(6);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertValuesOnly(2, 3, 5, 6);
+    }
+
+    @Test
+    public void testDeliverLatestToView_SingleItemViewComesAndGoes() throws Exception {
+        mPresenter.create();
+
+        PublishSubject<Integer> source = PublishSubject.create();
+        TestObserver<Integer> testObserver = new TestObserver<>();
+
+        source
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        source.onNext(1);
+        source.onNext(2);
+        mPresenter.attachView(mView);
+        mPresenter.detachView();
+        mPresenter.attachView(mView);
+        mPresenter.detachView();
+        mPresenter.attachView(mView);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertValuesOnly(2, 2, 2);
+    }
+
+    @Test
+    public void testDeliverLatestToView_Empty() throws Exception {
+        mPresenter.create();
+
+        TestObserver<Integer> testObserver = new TestObserver<>();
+        Observable.<Integer>empty()
+                .compose(RxTiPresenterUtils.<Integer>deliverLatestToView(mPresenter))
+                .subscribe(testObserver);
+
+        mPresenter.attachView(mView);
+
+        testObserver.assertNotComplete();
+        testObserver.assertNoErrors();
+        testObserver.assertEmpty();
+    }
+
+}