BaseRxFragment.java

package com.buggycoder.android.common.ui.frag;

import android.app.Activity;
import android.app.Fragment;
import android.content.Context;
import android.os.Bundle;

import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.android.app.AppObservable;
import timber.log.Timber;

public abstract class BaseRxFragment<T, R> extends Fragment {
    private static final String TAG = BaseRxFragment.class.getSimpleName();
    private static final int INITIAL_SUBSCRIBERS = 2;

    private List<Subscription> mSubscriptions;
    private ProxySubscriber mSourceSubscriber;

    public BaseRxFragment() {
        super();
        setRetainInstance(true);
    }

    @SuppressWarnings("unchecked")
    @Override
    public void onAttach(Activity activity) {
        super.onAttach(activity);

        if (activity instanceof RequestListener) {
            if (mSourceSubscriber == null) {
                mSourceSubscriber = new ProxySubscriber();
            }

            try {
                mSourceSubscriber.setRequestListener((RequestListener<R>) activity);
            } catch (ClassCastException e) {
                Timber.e(e, TAG);
                mSourceSubscriber = null;
            }
        }
    }

    @Override
    public void onDetach() {
        super.onDetach();

        if (mSourceSubscriber != null) {
            mSourceSubscriber.setRequestListener(null);
        }
    }

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        Bundle args = getArguments();
        if (!processArguments(args)) {
            //throw new IllegalArgumentException("Fragment received invalid arguments");

            Timber.w("Fragment received invalid arguments");
            return;
        }

        mSubscriptions = new ArrayList<>(INITIAL_SUBSCRIBERS);

        Context context = getActivity().getApplicationContext();

        Observable<T> sourceObservable = createObservable(context, args);
        if (sourceObservable != null) {
            Observable<T> uiSourceObservable = AppObservable.bindFragment(this, sourceObservable);
            if (mSourceSubscriber != null) {
                mSubscriptions.add(uiSourceObservable.subscribe(mSourceSubscriber));
            }

            Subscriber<T> uiSubscriber = createSubscriber(context, args);
            if (uiSubscriber != null) {
                mSubscriptions.add(uiSourceObservable.subscribe(uiSubscriber));
            }
        }
    }

    @Override
    public void onActivityCreated(Bundle savedInstanceState) {
        super.onActivityCreated(savedInstanceState);

        if (mSourceSubscriber != null) {
            mSourceSubscriber.notifyListener();
        }
    }


    @Override
    public void onDestroy() {
        unsubscribe();
        super.onDestroy();
    }


    public boolean isComplete() {
        return (mSourceSubscriber == null || mSourceSubscriber.isComplete());
    }

    public void unsubscribe() {
        if (mSubscriptions != null) {
            for (Subscription s : mSubscriptions) {
                if (!s.isUnsubscribed()) {
                    s.unsubscribe();
                }
            }
        }
    }


    /**
     * Process intermediate items of type T emitted by the Observable, mainly progress updates.
     * @param t item emitted by the connected Observable.
     * @return int value denoted progress percentage.
     */
    protected int processProgress(T t) {
        return 0;
    }


    protected Subscriber<T> createSubscriber(Context context, Bundle args) {
        // in case a child would like to tap into the Observable
        return null;
    }

    /**
     * Process input arguments
     * @param args Arguments received from the activity that is creating this fragment.
     * @return true if arguments supplied were sufficient, else false
     */
    protected abstract boolean processArguments(Bundle args);


    /**
     * Classes implementing this method should create and return the Observable
     * to which this fragment should subscribe to.
     * should subscribe.
     * @param context App (preferred) or Activity context
     * @param args Arguments given for creating this fragment
     * @return Observable to subscribe to, or null.
     */
    protected abstract Observable<T> createObservable(Context context, Bundle args);


    /**
     * Perform any transformations on type T (returned by the Observable) to type R.
     * e.g. searchResult.listings.get(0) for SearchResult -> Listing
     * @param t Object received from the Observable result.
     * @return Transformed object.
     */
    protected abstract R transformResult(T t);


    /**
     * Interface to be implemented by Activities interested in the Observable results.
     * @param <R> Type for the main result
     */
    public interface RequestListener<R> {
        void onProgress(int progress);

        void onError(Throwable e);

        void onSuccess(R data);
    }


    /**
     * Rx Subscriber that passes results of an Observable to the set RequestListener, if any.
     * Interested Activities are expected to implement the RequestListener interface, and will
     * automatically receive results from the Observable when this Fragment is attached.
     */
    private class ProxySubscriber extends Subscriber<T> {

        private int mProgress;
        private R mResult;
        private Throwable mException;
        private boolean mIsComplete;

        private RequestListener<R> mRequestListener;

        public ProxySubscriber() {
        }

        @Override
        public void onCompleted() {
            mIsComplete = true;
        }

        @Override
        public void onError(Throwable e) {
            mException = e;
            notifyListener();
        }

        @Override
        public void onNext(T t) {
            mProgress = processProgress(t);
            mResult = transformResult(t);
            notifyListener();
        }

        public void notifyListener() {
            if (mRequestListener != null) {
                // Error \/ Success
                if (mException != null) {
                    mRequestListener.onError(mException);
                } else {
                    if (mResult != null) {
                        // We already have the result
                        mRequestListener.onSuccess(mResult);
                    } else if (mProgress > 0) {
                        // We're in progress
                        mRequestListener.onProgress(mProgress);
                    }
                }
            }
        }

        public RequestListener<R> getRequestListener() {
            return mRequestListener;
        }

        public void setRequestListener(RequestListener<R> requestListener) {
            mRequestListener = requestListener;
        }

        public int getProgress() {
            return mProgress;
        }

        public R getResult() {
            return mResult;
        }

        public boolean hasError() {
            return (mException != null);
        }

        public Throwable getException() {
            return mException;
        }

        public boolean isComplete() {
            return mIsComplete;
        }
    }
}