class LiveDataReactiveStreams {

    public static <T> LiveData<T> fromPublisher(final Publisher<T> publisher) {
        MutableLiveData<T> liveData = new MutableLiveData<>();
        // Since we don't have a way to directly observe cancels, weakly hold the live data.
        final WeakReference<MutableLiveData<T>> liveDataRef = new WeakReference<>(liveData);

        publisher.subscribe(new Subscriber<T>() {
            @Override
            public void onSubscribe(Subscription s) {
                // Don't worry about backpressure. If the stream is too noisy then backpressure can
                // be handled upstream.
                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) {
                // Errors should be handled upstream, so propagate as a crash.
                throw new RuntimeException(t);
            }

            @Override
            public void onComplete() {
            }
        });

        return liveData;
    }

    private  LiveDataReactiveStreams();

}