LiveDataReactiveStreams { @NonNull public static <T> LiveData<T> fromPublisher(@NonNull Publisher<T> publisher) { return new PublisherLiveData<>(publisher); } private  LiveDataReactiveStreams(); @NonNull static Publisher<T> toPublisher(
            @NonNull LifecycleOwner lifecycle, @NonNull LiveData<T> liveData); @NonNull static LiveData<T> fromPublisher(@NonNull Publisher<T> publisher);  }
@Test public void convertsFromPublisher() { PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("foo"); processor.onNext("bar"); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar", "baz"))); }
@Test public void convertsFromPublisherSubscribeWithDelay() { PublishProcessor<String> processor = PublishProcessor.create(); processor.delaySubscription(100, TimeUnit.SECONDS, sBackgroundScheduler); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("foo"); liveData.removeObserver(mObserver); sBackgroundScheduler.triggerActions(); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("bar"); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "foo", "bar", "baz"))); }
@Test public void convertsFromPublisherThrowsException() { PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); IllegalStateException exception = new IllegalStateException("test exception"); try { processor.onError(exception); fail("Runtime Exception expected"); } catch (RuntimeException ex) { assertEquals(ex.getCause(), exception); } }
@Test public void convertsFromPublisherWithMultipleObservers() { final List<String> output2 = new ArrayList<>(); PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("foo"); processor.onNext("bar"); liveData.observe(mLifecycleOwner, new Observer<String>() { @Override public void onChanged(@Nullable String s) { output2.add(s); } }); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar", "baz"))); assertThat(output2, is(Arrays.asList("bar", "baz"))); }
@Test public void convertsFromPublisherWithMultipleObserversAfterInactive() { final List<String> output2 = new ArrayList<>(); PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("foo"); processor.onNext("bar"); liveData.observe(mLifecycleOwner, new Observer<String>() { @Override public void onChanged(@Nullable String s) { output2.add(s); } }); liveData.removeObserver(mObserver); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar"))); assertThat(output2, is(Arrays.asList("bar", "baz"))); }
@Test public void convertsFromPublisherAfterInactive() { PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("foo"); liveData.removeObserver(mObserver); processor.onNext("bar"); liveData.observe(mLifecycleOwner, mObserver); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "foo", "baz"))); }
@Test public void convertsFromPublisherManagesSubscriptions() { PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); assertThat(processor.hasSubscribers(), is(false)); liveData.observe(mLifecycleOwner, mObserver); assertThat(processor.hasSubscribers(), is(true)); liveData.removeObserver(mObserver); assertThat(processor.hasSubscribers(), is(false)); }
@Test public void convertsFromAsyncPublisher() { Flowable<String> input = Flowable.just("foo") .concatWith(Flowable.just("bar", "baz").observeOn(sBackgroundScheduler)); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(input); liveData.observe(mLifecycleOwner, mObserver); assertThat(mLiveDataOutput, is(Collections.singletonList("foo"))); sBackgroundScheduler.triggerActions(); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar", "baz"))); }