Step { protected Observable<Optional<Data<T, A>>> asObservable() { final Observable<Optional<Data<T, A>>> cachedObservable = stepDataSingle.toObservable().observeOn(AndroidSchedulers.mainThread()).cache(); return cachedObservable.flatMap( new Function<Optional<Data<T, A>>, ObservableSource<Optional<Data<T, A>>>>() { @Override public ObservableSource<Optional<Data<T, A>>> apply(Optional<Data<T, A>> dataOptional) throws Exception { if (dataOptional.isPresent()) { A actionableItem = dataOptional.get().actionableItem; return actionableItem .lifecycle() .filter( new Predicate<InteractorEvent>() { @Override public boolean test(InteractorEvent interactorEvent) throws Exception { return interactorEvent == InteractorEvent.ACTIVE; } }) .zipWith( cachedObservable, new BiFunction< InteractorEvent, Optional<Data<T, A>>, Optional<Data<T, A>>>() { @Override public Optional<Data<T, A>> apply( InteractorEvent interactorEvent, Optional<Data<T, A>> dataOptional) throws Exception { return dataOptional; } }); } else { return Observable.just(Optional.<Data<T, A>>absent()); } } }); } private  Step(Single<Optional<Data<T, A>>> stepDataSingle); static Step<T, A> from(Single<Data<T, A>> stepDataSingle); static Step<T, A> fromOptional(
      Single<Optional<Data<T, A>>> stepDataSingle); Step<TNewValueType, TNewActionableItem> onStep(
          final BiFunction<T, A, Step<TNewValueType, TNewActionableItem>> func);  }
@Test public void asObservable_withInactiveLifecycle_shouldWaitForActiveLifecycleBeforeEmitting() { Object returnValue = new Object(); TestObserver<Optional<Step.Data<Object, ActionableItem>>> testSubscriber = new TestObserver<>(); step.asObservable().subscribe(testSubscriber); testSubscriber.assertNoValues(); testSubscriber.assertNoErrors(); testSubscriber.assertNotComplete(); returnValueSubject.onNext( Optional.of( new Step.Data<Object, ActionableItem>( returnValue, new ActionableItem() { @NonNull @Override public Observable<InteractorEvent> lifecycle() { return interactorLifecycleSubject; } }))); returnValueSubject.onComplete(); testSubscriber.assertNoValues(); testSubscriber.assertNoErrors(); testSubscriber.assertNotComplete(); interactorLifecycleSubject.onNext(InteractorEvent.ACTIVE); testSubscriber.assertValueCount(1); assertThat(testSubscriber.values().get(0).get().getValue()).isEqualTo(returnValue); testSubscriber.assertComplete(); testSubscriber.assertNoErrors(); }
@Test public void asObservable_withActiveLifecycle_shouldEmitWithoutWaiting() { Object returnValue = new Object(); TestObserver<Optional<Step.Data<Object, ActionableItem>>> testSubscriber = new TestObserver<>(); interactorLifecycleSubject.onNext(InteractorEvent.ACTIVE); step.asObservable().subscribe(testSubscriber); testSubscriber.assertNoValues(); testSubscriber.assertNoErrors(); testSubscriber.assertNotComplete(); returnValueSubject.onNext( Optional.of( new Step.Data<Object, ActionableItem>( returnValue, new ActionableItem() { @NonNull @Override public Observable<InteractorEvent> lifecycle() { return interactorLifecycleSubject; } }))); returnValueSubject.onComplete(); testSubscriber.assertValueCount(1); assertThat(testSubscriber.values().get(0).get().getValue()).isEqualTo(returnValue); testSubscriber.assertComplete(); testSubscriber.assertNoErrors(); }