LiveDataReactiveStreams { public static <T> LiveData<T> fromPublisher(final Publisher<T> publisher) { MutableLiveData<T> liveData = new MutableLiveData<>(); final WeakReference<MutableLiveData<T>> liveDataRef = new WeakReference<>(liveData); publisher.subscribe(new Subscriber<T>() { @Override public void onSubscribe(Subscription s) { s.request(Long.MAX_VALUE); } @Override public void onNext(final T t) { final LiveData<T> liveData = liveDataRef.get(); if (liveData != null) { liveData.postValue(t); } } @Override public void onError(Throwable t) { throw new RuntimeException(t); } @Override public void onComplete() { } }); return liveData; } private  LiveDataReactiveStreams(); static Publisher<T> toPublisher(
            final LifecycleOwner lifecycle, final LiveData<T> liveData); static LiveData<T> fromPublisher(final Publisher<T> publisher);  }
@Test public void convertsFromPublisher() { PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(S_LIFECYCLE_OWNER, mObserver); processor.onNext("foo"); processor.onNext("bar"); processor.onNext("baz"); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar", "baz"))); }
@Test public void convertsFromPublisherWithMultipleObservers() { final List<String> output2 = new ArrayList<>(); PublishProcessor<String> processor = PublishProcessor.create(); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(processor); liveData.observe(S_LIFECYCLE_OWNER, mObserver); processor.onNext("foo"); processor.onNext("bar"); liveData.observe(S_LIFECYCLE_OWNER, 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 convertsFromAsyncPublisher() { Flowable<String> input = Flowable.just("foo") .concatWith(Flowable.just("bar", "baz").observeOn(sBackgroundScheduler)); LiveData<String> liveData = LiveDataReactiveStreams.fromPublisher(input); liveData.observe(S_LIFECYCLE_OWNER, mObserver); assertThat(mLiveDataOutput, is(Collections.singletonList("foo"))); sBackgroundScheduler.triggerActions(); assertThat(mLiveDataOutput, is(Arrays.asList("foo", "bar", "baz"))); }