RxJava1.x 升级到 RxJava2.x

0、RxApp项目 Rxjava1.x 升级到Rxjava2.x

RxApp项目地址，但是这个项目的Rxjava是1.+版本的，但是RxJava 1.x版本支持到2018年1月1日，很显然项目用到rxjava的地方必须升级到2.x。

1、Rxjava1.x 和 Rxjava2.x 区别

网上很多介绍区别的，主要是讲解2.x版本支持背压，具体看我搜到的文章，总结下来就是：
Flowable 支持背压、效率低
Observable 不支持背压、效率高
当然上面只是笼统的说了一下，具体需要看上面的文章。

2、升级的问题

下面要说的是，RxApp中封装了Rxjava+Retrofit统一处理了网络返回的数据，就像这篇文章所说的那样，统一处理了返回数据、统一增加缓存、统一切换线程、统一管理生命周期等优点，比如使用CompositeSubscription来收集Subscription，来统一取消订阅，按照上面所说的要把Observable升级到2.x，但是在RxJava2.0中，由于subscribe()方法现在返回void，那怎么办呢？
其实在RxJava2.0中，Flowable提供了subscribeWith这个方法可以返回当前订阅的观察者，并且通过ResourceSubscriber DisposableSubscriber等观察者来提供 Disposable的接口，这样升级到2.x还是能保持上面所有的优点！！
下面是部分代码，具体代码看github项目地址。

public class RxUtils {
    private static FlowableTransformer ioToMainThreadSchedulerTransformer;
    private static FlowableTransformer newThreadToMainThreadSchedulerTransformer;

    static {
        ioToMainThreadSchedulerTransformer = createIOToMainThreadScheduler();
        newThreadToMainThreadSchedulerTransformer = createNewThreadToMainThreadScheduler();
    }

    private static <T> FlowableTransformer<T, T> createIOToMainThreadScheduler() {
        return tObservable -> tObservable.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.computation())
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 从IO线程切换到主线程
     * @param <T>
     * @return
     */
    public static <T> FlowableTransformer<T, T> applyIOToMainThreadSchedulers() {
        return ioToMainThreadSchedulerTransformer;
    }

    private static <T> FlowableTransformer<T, T> createNewThreadToMainThreadScheduler() {
        return tObservable -> tObservable.subscribeOn(Schedulers.newThread())
                .unsubscribeOn(Schedulers.computation())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    public static <T> FlowableTransformer<T, T> applyNewThreadToMainThreadSchedulers() {
        return newThreadToMainThreadSchedulerTransformer;
    }

    /**
     * 处理服务器返回的数据，进一步处理错误信息
     * @param <T>
     * @return
     */
    public static <T> FlowableTransformer<BaseBean<T>, T> handleResult(){
        return new FlowableTransformer<BaseBean<T>, T>() {
            @Override
            public Publisher<T> apply(@NonNull Flowable<BaseBean<T>> flowable) {
                return flowable.flatMap(new Function<BaseBean<T>, Publisher<T>>() {
                    @Override
                    public Publisher<T> apply(@NonNull BaseBean<T> tBaseBean) throws Exception {
                        if (!tBaseBean.isError()){
                            return createData(tBaseBean.getResults());
                        }else{
                            return Flowable.error(new ServerException("服务器返回错误"));
                        }
                    }
                }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
            }
        };

    }

    /**
     * 创建成功的数据
     *
     * @param data
     * @param <T>
     * @return
     */
    private static <T> Flowable<T> createData(T data) {
        return Flowable.create(new FlowableOnSubscribe<T>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<T> flowableEmitter) throws Exception {
                try {
                    flowableEmitter.onNext(data);
                    flowableEmitter.onComplete();
                }catch (Exception e){
                    flowableEmitter.onError(e);
                }
            }
        }, BackpressureStrategy.BUFFER);
    }

    /**
     * 自定义 服务器返回异常
     */
    public static class ServerException extends Throwable {
        public ServerException(String msg) {}
    }

}

public abstract class RxSubUtils<T> extends DisposableSubscriber<T> {
    private CompositeDisposable compositeDisposable;
    private Context mContext;
    private String msg;

    public RxSubUtils(CompositeDisposable mCompositeSubscription) {
        this.compositeDisposable = mCompositeSubscription;
    }

    /**
     * @param context context
     * @param msg     dialog message
     */
    public RxSubUtils(CompositeDisposable mCompositeSubscription, Context context, String msg) {
        this.compositeDisposable = mCompositeSubscription;
        this.mContext = context;
        this.msg = msg;
    }

    /**
     * @param context context
     */
    public RxSubUtils(CompositeDisposable mCompositeSubscription, Context context) {
        this(mCompositeSubscription, context, "请稍后...");
    }

    /**
     * 这个一定要有 Presenter的逻辑在这里处理
     * @param t
     */
    @Override
    public void onNext(T t) {
        _onNext(t);
    }

    @Override
    public void onError(Throwable e) {
        e.printStackTrace();
        LoadingDialogManager.getLoadingDialog().hideDialog();

        if (!NetWorkUtils.isNetworkAvailable()) {
            ToastUtil.show(R.string.net_error);
        } else if (e instanceof RxUtils.ServerException) {
//            String s = ((RxUtils.ServerException)e).getMsg();
//            //token 过期了
//            if(TextUtils.equals(s, RxUtils.TOKEN_OVER_TIME) && mContext != null){
//                Utils.startLoginActivity(mContext);
//            }
//            ToastUtil.show(s);
        } else {
            ToastUtil.show(R.string.error);
        }

        _onError();
    }

    @Override
    public void onComplete() {
        if (compositeDisposable != null)
            compositeDisposable.clear();

        LoadingDialogManager.getLoadingDialog().hideDialog();
    }

    @Override
    public void onStart() {
        super.onStart();
        if (mContext != null) {
            LoadingDialogManager.getLoadingDialog().showDialog(mContext);
        }
    }

    protected abstract void _onNext(T t);

    /**
     * 错误处理，需要的话重写这个方法
     */
    protected void _onError(){

    }

}