RxUi { public static <T> Function<Observable<T>, Disposable> ui(final Consumer<T> uiAction) { return new Function<Observable<T>, Disposable>() { @Override public Disposable apply(Observable<T> observable) { return observable .observeOn(mainThread()) .subscribe(uiAction); } }; }  static Disposable bind(Observable<T> observable, Function<Observable<T>, Disposable> uiFunc); static Function<Observable<T>, Disposable> ui(final Consumer<T> uiAction);  }
@Test public void uiShouldNotCallActionUntilMainThreadIsPaused() throws Exception { Consumer<String> uiAction = mock(Consumer.class); Function<Observable<String>, Disposable> binderFunc = RxUi.ui(uiAction); ShadowLooper.pauseMainLooper(); binderFunc.apply(Observable.just("a", "b", "c")); verifyZeroInteractions(uiAction); }
@Test public void uiShouldCallAction() throws Exception { final AtomicBoolean wasCalledOnUiThread = new AtomicBoolean(true); Consumer<String> uiAction = spy(new Consumer<String>() { @Override public void accept(String s) { if (!wasCalledOnUiThread.compareAndSet(true, Looper.getMainLooper() == Looper.myLooper())) { throw new IllegalStateException("Not on Main Thread!"); } } }); InOrder inOrder = inOrder(uiAction); Function<Observable<String>, Disposable> binderFunc = RxUi.ui(uiAction); binderFunc.apply(Observable.just("a", "b", "c")); inOrder.verify(uiAction).accept("a"); inOrder.verify(uiAction).accept("b"); inOrder.verify(uiAction).accept("c"); inOrder.verifyNoMoreInteractions(); assertThat(wasCalledOnUiThread.get()).isTrue(); }
@Test public void uiShouldPreventCallsToActionIfUnsubscribedBeforeExecution() throws Exception { Consumer<String> uiAction = mock(Consumer.class); Function<Observable<String>, Disposable> binderFunc = RxUi.ui(uiAction); ShadowLooper.pauseMainLooper(); Disposable disposable = binderFunc.apply(Observable.just("a", "b", "c")); disposable.dispose(); ShadowLooper.runUiThreadTasksIncludingDelayedTasks(); verifyZeroInteractions(uiAction); }