model 中添加获取网络数据的管理

创建管理容器

/**
  * 将所有正在处理的Subscription都添加到CompositeSubscription中。统一退出的时候注销观察
  */
 private CompositeDisposable mCompositeDisposable;

将网络请求添加到容器中

    /**
     * 将网络请求的每一个disposable添加进入CompositeDisposable，再退出时候一并注销
     *
     * @param subscription
     */
    private void addDisposable(Disposable subscription) {
        //csb 如果解绑了的话添加 sb 需要新的实例否则绑定时无效的
        if (mCompositeDisposable == null || mCompositeDisposable.isDisposed()) {
            mCompositeDisposable = new CompositeDisposable();
        }
        mCompositeDisposable.add(subscription);
    }

在页面销毁时注销所有请求

  /**
     * 在页面销毁时注销所有请求
     */
    private void unDisposable() {
        if (mCompositeDisposable != null) {
            mCompositeDisposable.dispose();
        }
    }

数据返回基类

    /**
     * 数据返回基类
     */
    public class BaseResults {
        int code;
        String message;
    }

添加简单数据信息过滤 ; 添加到网络请求容器中 ; 添加异步调度器

 /**
     * 添加简单数据信息过滤
     * 添加到网络请求容器中
     * 添加异步调度器
     *
     * @param <R>
     * @return
     */
    public <R extends BaseResults> ObservableTransformer<R, R> applySchedulers() {
        return new ObservableTransformer<R, R>() {
            @Override
            public ObservableSource<R> apply(@NonNull Observable<R> upstream) {
                return upstream
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                /**
                                 * 添加请求管理
                                 */
                                addDisposable(disposable);
                            }
                        })
                        .map(new Function<R, R>() {
                            @Override
                            public R apply(@NonNull R r) throws Exception {
                                if (1 == r.code) {
                                    return r;
                                } else {
                                    throw new RuntimeException(r.message);
                                }
                            }
                        })
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

测试

    /**
     * 测试
     */
    public void test() {
        RetrofitHelper.getInstence().test(new ArnoHeaders())
                .compose(this.<BaseResults>applySchedulers())
                .subscribe(new Consumer<BaseResults>() {
                    @Override
                    public void accept(BaseResults responseBody) throws Exception {
                        Log.i("====>", responseBody.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.i("====>", throwable.toString());
                    }
                });
    }