@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"))); }
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; }
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; } }
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(); }
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); }
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);  }