Architecture Components Retrofit and RxJava 2 error handling

Rather than exposing just the list of shows through your LiveData object you would need to wrap the shows and error into a class that can hold the error.

With your example you could do something like this:

    LiveDataReactiveStreams.fromPublisher(Observable.fromIterable(ShowsUtil.loadsIds())
            .subscribeOn(Schedulers.io())
            .flatMap(clientAdapter::getShowWithNextEpisode)
            .observeOn(Schedulers.computation())
            .toSortedList(new ShowsUtil.ShowComparator())
            .observeOn(AndroidSchedulers.mainThread())
            .map(Result::success)
            .onErrorReturn(Result::error)
            .toFlowable());

Where Result is the wrapper class that holds either the error or result

final class Result<T> {

    private final T result;
    private final Throwable error;

    private Result(@Nullable T result, @Nullable Throwable error) {
        this.result = result;
        this.error = error;
    }

    @NonNull
    public static <T> Result<T> success(@NonNull T result) {
        return new Result(result, null);
    }

    @NonNull
    public static <T> Result<T> error(@NonNull Throwable error) {
        return new Result(null, error);
    }

    @Nullable
    public T getResult() {
        return result;
    }

    @Nullable
    public Throwable getError() {
        return error;
    }
}

I have used a solution with LiveDataReactiveStreams.fromPublisher along with a generic wrapper class Result that was based on the following class.

https://github.com/googlesamples/android-architecture-components/blob/master/GithubBrowserSample/app/src/main/java/com/android/example/github/vo/Resource.kt

The idea is similar to https://stackoverflow.com/a/51669953/4266287

This pattern can be used regardless of using LiveData and the main idea is to be able to handle errors without ending the stream. So it works similar to RxRelay where even your errors will be emitted using onNext and the actual meaning comes from the data, whether the result or error are present.

If this will be part of the backbone of your project, you can create some extra framework to make all this transparent. For example, you can create your own class that would be something like a Resource<T> aware stream that has an observe method that mimics the way you call RxJava's observe but "unwrap" the data and call the right callback:

fun subscribe(
        onNext: (T) -> Unit,
        onError: ((Throwable) -> Unit)? = null
) {
    val handleOnNext: (Resource<T>) -> Unit = { resource: Resource<T> ->
        when (resource.status) {
            Status.SUCCESS -> resource.data?.let(onNext)
            Status.ERROR -> resource.error?.let { onError?.invoke(it) ?: throw it }
        }
    }

    publishSubject
        .subscribeOn(subscribeOn)
        .observeOn(observeOn)
        .run { subscribe(handleOnNext) }
        .addTo(compositeDisposable)
}

In my case I copied LiveDataReactiveStreams.java into my 'util' package as 'MyLiveDataReactiveStreams.java'. The modified version replaces the RuntimeException with a GreenRobot EventBus post. Then in my app I can subscribe to that event and handle the error appropriately. For this solution I had to add '@SuppressLint("RestrictedApi")' in 3 places. I am not sure if Google allows that for play store apps. This repo contains a complete example: https://github.com/LeeHounshell/Dogs Below is the relevant code:

    // add GreenRobot to your app/build.gradle

    def greenrobot_version = '3.2.0'
    def timberkt_version = '1.5.1'

    implementation "org.greenrobot:eventbus:$greenrobot_version"
    implementation "com.github.ajalt:timberkt:$timberkt_version"


//--------------------------------------------------
// next put this in your Activity or Fragment to handle the error

override fun onStart() {
    super.onStart()
    EventBus.getDefault().register(this)
}

override fun onStop() {
    super.onStop()
    EventBus.getDefault().unregister(this);
}

@Subscribe(threadMode = ThreadMode.MAIN)
fun onRxErrorEvent(rxError_event: RxErrorEvent) {
    // do your custom error handling here
    Toast.makeText(activity, rxError_event.errorDescription, Toast.LENGTH_LONG).show()
}



//--------------------------------------------------
// create a new class in 'util' to post the error

import com.github.ajalt.timberkt.Timber
import org.greenrobot.eventbus.EventBus

class RxErrorEvent(val description: String) {
    private val _tag = "LEE: <" + RxErrorEvent::class.java.simpleName + ">"

    lateinit var errorDescription: String

    init {
        errorDescription = description
    }

    fun post() {
        Timber.tag(_tag).e("post $errorDescription")
        EventBus.getDefault().post(this)
    }

}


//--------------------------------------------------
// and use MyLiveDataReactiveStreams (below) instead of LiveDataReactiveStreams

/*
 *  This is a modified version of androidx.lifecycle.LiveDataReactiveStreams
 *  The original LiveDataReactiveStreams object can't handle RxJava error conditions.
 *  Now errors are emitted as RxErrorEvent objects using the GreenRobot EventBus.
 */

import android.annotation.SuppressLint;

import androidx.annotation.NonNull;
import androidx.annotation.Nullable;
import androidx.arch.core.executor.ArchTaskExecutor;
import androidx.lifecycle.LifecycleOwner;
import androidx.lifecycle.LiveData;
import androidx.lifecycle.Observer;

import com.github.ajalt.timberkt.Timber;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.atomic.AtomicReference;

/**
 * Adapts {@link LiveData} input and output to the ReactiveStreams spec.
 */
@SuppressWarnings("WeakerAccess")
public final class MyLiveDataReactiveStreams {
    private final static String _tag = "LEE: <" + MyLiveDataReactiveStreams.class.getSimpleName() + ">";

    private MyLiveDataReactiveStreams() {
    }

    /**
     * Adapts the given {@link LiveData} stream to a ReactiveStreams {@link Publisher}.
     *
     * <p>
     * By using a good publisher implementation such as RxJava 2.x Flowables, most consumers will
     * be able to let the library deal with backpressure using operators and not need to worry about
     * ever manually calling {@link Subscription#request}.
     *
     * <p>
     * On subscription to the publisher, the observer will attach to the given {@link LiveData}.
     * Once {@link Subscription#request} is called on the subscription object, an observer will be
     * connected to the data stream. Calling request(Long.MAX_VALUE) is equivalent to creating an
     * unbounded stream with no backpressure. If request with a finite count reaches 0, the observer
     * will buffer the latest item and emit it to the subscriber when data is again requested. Any
     * other items emitted during the time there was no backpressure requested will be dropped.
     */
    @NonNull
    public static <T> Publisher<T> toPublisher(
            @NonNull LifecycleOwner lifecycle, @NonNull LiveData<T> liveData) {

        return new MyLiveDataReactiveStreams.LiveDataPublisher<>(lifecycle, liveData);
    }

    private static final class LiveDataPublisher<T> implements Publisher<T> {
        final LifecycleOwner mLifecycle;
        final LiveData<T> mLiveData;

        LiveDataPublisher(LifecycleOwner lifecycle, LiveData<T> liveData) {
            this.mLifecycle = lifecycle;
            this.mLiveData = liveData;
        }

        @Override
        public void subscribe(Subscriber<? super T> subscriber) {
            subscriber.onSubscribe(new MyLiveDataReactiveStreams.LiveDataPublisher.LiveDataSubscription<T>(subscriber, mLifecycle, mLiveData));
        }

        static final class LiveDataSubscription<T> implements Subscription, Observer<T> {
            final Subscriber<? super T> mSubscriber;
            final LifecycleOwner mLifecycle;
            final LiveData<T> mLiveData;

            volatile boolean mCanceled;
            // used on main thread only
            boolean mObserving;
            long mRequested;
            // used on main thread only
            @Nullable
            T mLatest;

            LiveDataSubscription(final Subscriber<? super T> subscriber,
                                 final LifecycleOwner lifecycle, final LiveData<T> liveData) {
                this.mSubscriber = subscriber;
                this.mLifecycle = lifecycle;
                this.mLiveData = liveData;
            }

            @Override
            public void onChanged(@Nullable T t) {
                if (mCanceled) {
                    return;
                }
                if (mRequested > 0) {
                    mLatest = null;
                    mSubscriber.onNext(t);
                    if (mRequested != Long.MAX_VALUE) {
                        mRequested--;
                    }
                } else {
                    mLatest = t;
                }
            }

            @SuppressLint("RestrictedApi")
            @Override
            public void request(final long n) {
                if (mCanceled) {
                    return;
                }
                ArchTaskExecutor.getInstance().executeOnMainThread(new Runnable() {
                    @Override
                    public void run() {
                        if (mCanceled) {
                            return;
                        }
                        if (n <= 0L) {
                            mCanceled = true;
                            if (mObserving) {
                                mLiveData.removeObserver(MyLiveDataReactiveStreams.LiveDataPublisher.LiveDataSubscription.this);
                                mObserving = false;
                            }
                            mLatest = null;
                            mSubscriber.onError(
                                    new IllegalArgumentException("Non-positive request"));
                            return;
                        }

                        // Prevent overflowage.
                        mRequested = mRequested + n >= mRequested
                                ? mRequested + n : Long.MAX_VALUE;
                        if (!mObserving) {
                            mObserving = true;
                            mLiveData.observe(mLifecycle, MyLiveDataReactiveStreams.LiveDataPublisher.LiveDataSubscription.this);
                        } else if (mLatest != null) {
                            onChanged(mLatest);
                            mLatest = null;
                        }
                    }
                });
            }

            @SuppressLint("RestrictedApi")
            @Override
            public void cancel() {
                if (mCanceled) {
                    return;
                }
                mCanceled = true;
                ArchTaskExecutor.getInstance().executeOnMainThread(new Runnable() {
                    @Override
                    public void run() {
                        if (mObserving) {
                            mLiveData.removeObserver(MyLiveDataReactiveStreams.LiveDataPublisher.LiveDataSubscription.this);
                            mObserving = false;
                        }
                        mLatest = null;
                    }
                });
            }
        }
    }

    /**
     * Creates an observable {@link LiveData} stream from a ReactiveStreams {@link Publisher}}.
     *
     * <p>
     * When the LiveData becomes active, it subscribes to the emissions from the Publisher.
     *
     * <p>
     * When the LiveData becomes inactive, the subscription is cleared.
     * LiveData holds the last value emitted by the Publisher when the LiveData was active.
     * <p>
     * Therefore, in the case of a hot RxJava Observable, when a new LiveData {@link Observer} is
     * added, it will automatically notify with the last value held in LiveData,
     * which might not be the last value emitted by the Publisher.
     * <p>
     * Note that LiveData does NOT handle errors and it expects that errors are treated as states
     * in the data that's held. In case of an error being emitted by the publisher, an error will
     * be propagated to the main thread and the app will crash.
     *
     * @param <T> The type of data hold by this instance.
     */
    @NonNull
    public static <T> LiveData<T> fromPublisher(@NonNull Publisher<T> publisher) {
        return new MyLiveDataReactiveStreams.PublisherLiveData<>(publisher);
    }

    /**
     * Defines a {@link LiveData} object that wraps a {@link Publisher}.
     *
     * <p>
     * When the LiveData becomes active, it subscribes to the emissions from the Publisher.
     *
     * <p>
     * When the LiveData becomes inactive, the subscription is cleared.
     * LiveData holds the last value emitted by the Publisher when the LiveData was active.
     * <p>
     * Therefore, in the case of a hot RxJava Observable, when a new LiveData {@link Observer} is
     * added, it will automatically notify with the last value held in LiveData,
     * which might not be the last value emitted by the Publisher.
     *
     * <p>
     * Note that LiveData does NOT handle errors and it expects that errors are treated as states
     * in the data that's held. In case of an error being emitted by the publisher, an error will
     * be propagated to the main thread and the app will crash.
     *
     * @param <T> The type of data hold by this instance.
     */
    private static class PublisherLiveData<T> extends LiveData<T> {
        private final Publisher<T> mPublisher;
        final AtomicReference<MyLiveDataReactiveStreams.PublisherLiveData.LiveDataSubscriber> mSubscriber;

        PublisherLiveData(@NonNull Publisher<T> publisher) {
            mPublisher = publisher;
            mSubscriber = new AtomicReference<>();
        }

        @Override
        protected void onActive() {
            super.onActive();
            MyLiveDataReactiveStreams.PublisherLiveData.LiveDataSubscriber s = new MyLiveDataReactiveStreams.PublisherLiveData.LiveDataSubscriber();
            mSubscriber.set(s);
            mPublisher.subscribe(s);
        }

        @Override
        protected void onInactive() {
            super.onInactive();
            MyLiveDataReactiveStreams.PublisherLiveData.LiveDataSubscriber s = mSubscriber.getAndSet(null);
            if (s != null) {
                s.cancelSubscription();
            }
        }

        final class LiveDataSubscriber extends AtomicReference<Subscription>
                implements Subscriber<T> {

            @Override
            public void onSubscribe(Subscription s) {
                if (compareAndSet(null, s)) {
                    s.request(Long.MAX_VALUE);
                } else {
                    s.cancel();
                }
            }

            @Override
            public void onNext(T item) {
                postValue(item);
            }

            @SuppressLint("RestrictedApi")
            @Override
            public void onError(final Throwable ex) {
                mSubscriber.compareAndSet(this, null);

                ArchTaskExecutor.getInstance().executeOnMainThread(new Runnable() {
                    @Override
                    public void run() {
                        //NOTE: Errors are be handled upstream
                        Timber.tag(_tag).e("LiveData does not handle errors. Errors from publishers are handled upstream via EventBus. error: " + ex);
                        RxErrorEvent rx_event = new RxErrorEvent(ex.toString());
                        rx_event.post();
                    }
                });
            }

            @Override
            public void onComplete() {
                mSubscriber.compareAndSet(this, null);
            }

            public void cancelSubscription() {
                Subscription s = get();
                if (s != null) {
                    s.cancel();
                }
            }
        }
    }
}