Rxjava Demo

import cn.com.sxit.android.ydac.bean.app.request.AppVo;
import cn.com.sxit.android.ydac.bean.app.response.TappVersion;
import cn.com.sxit.android.ydac.bean.auth.request.LoginVo;
import cn.com.sxit.android.ydac.bean.auth.response.LoginSuccessDto;
import cn.com.sxit.android.ydac.net.Counter;
import cn.com.sxit.android.ydac.net.RequestVo;
import cn.com.sxit.android.ydac.net.ResponseVo;
import com.ways2u.AppContext;
import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import org.junit.Test;
import validate.MyException;

/**
 * Created by huanglong on 2016/12/20.
 * 流程
 * 序列号是否重置 ---是---> 请求更新序号
 *       |                     |
 *       否                   成功
 *       |                     |
 *       |                     |
 *       -----------------------
 *                 |
 *          |-->尝试请求数据----成功---->结束
 *          |       |
 *          |      失败
 *          |       |
 *          --是--session过时
 *                  |
 *               其他错误-------------->结束
 */
public class HttpRequest {

    @Test
    public void testTwoRequest() {
        //每个api需要写的代码
        AppVo appVo = new AppVo();
        RequestVo<AppVo> requestVo = RequestVo.buildResquest("checkUpdate", false, 0l, appVo);
        Observable<ResponseVo<TappVersion>> sourceObservable =
                Environment.getInstance().netApi.checkUpdate(requestVo);

        //模拟3001，session过时
        /*sourceObservable = sourceObservable.map(new Function<ResponseVo<T>, ResponseVo<T>>() {
            @Override
            public ResponseVo<T> apply(ResponseVo<T> tResponseVo) throws Exception {
                tResponseVo.errcode = "3001";
                return tResponseVo;
            }
        })*/

        call(sourceObservable)
                //.subscribeOn(Schedulers.io())
                //.observeOn(Schedulers.newThread())
                .subscribe(new Observer<TappVersion>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(TappVersion value) {
                System.out.println("期望结果：" + value);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError:" + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }

    //公共方法

    public <T> Observable<T> call(final Observable<ResponseVo<T>> api) {
        //是否要获取序列号
        if (Counter.getInstance().needReset()) {
            Observable<ResponseVo<Integer>> observable = Environment.getInstance().netApi.getClientNum();
            Observable<Integer> integerObservable = getResponseVoObservable(observable);
            return integerObservable.concatMap(new Function<Integer, ObservableSource<T>>() {
                @Override
                public ObservableSource<T> apply(Integer integer) throws Exception {

                    System.out.println("getClientNum:" + integer);

                    return validateLogin(api);
                }
            });
        } else {
            return validateLogin(api);
        }
    }

    /* session 过时会自动登录 */
    public <T> Observable<T> validateLogin(final Observable<ResponseVo<T>> api) {

        final Observable<T> observableSource;
        observableSource = api.concatMap(new Function<ResponseVo<T>, ObservableSource<T>>() {
                    @Override
                    public ObservableSource<T> apply(final ResponseVo<T> responseVo) throws Exception {
                        //session 失效
                        if (responseVo.errcode != null && responseVo.errcode.equals("3001")) {
                            //重新登录
                            System.out.println("重新登录:" + responseVo);
                            throw new MyException(responseVo.errmsg, responseVo.errcode);
                            //return login(callback);
                        } else {
                            //返回正确处理对象
                            System.out.println("ok:" + responseVo);
                            return valedateCode(responseVo);
                        }
                    }
                })
                //缓存一下操作，不然会重复操作
                .cache();

        return observableSource.onErrorResumeNext(new Function<Throwable, ObservableSource<? extends T>>() {
            @Override
            public ObservableSource<? extends T> apply(Throwable throwable) throws Exception {
                //System.out.println("----"+throwable);
                if ((throwable instanceof MyException) && "3001".equals(((MyException) throwable).errcode)) {
                    //重新登录
                    return login(api);
                } else {
                    //不改变信号流向
                    return observableSource;
                }
            }
        });
    }

    //登录
    private <T> ObservableSource<T> login(final Observable<ResponseVo<T>> api) {
        LoginVo appVo = new LoginVo();
        appVo.loginName = "monkey";
        appVo.loginPwd = "123";
        appVo.roleType = 1;

        RequestVo<LoginVo> requestVo = RequestVo.buildResquest("userLogin", false, 0l, appVo);

        Observable<ResponseVo<LoginSuccessDto>> observable =
                Environment.getInstance().netApi.userLogin(requestVo);

        return observable.concatMap(new Function<ResponseVo<LoginSuccessDto>, ObservableSource<T>>() {
            @Override
            public ObservableSource<T> apply(final ResponseVo<LoginSuccessDto> responseVo) throws Exception {
                if (responseVo.errcode != null && responseVo.errcode.equals("0")) {

                    AppContext.getInstance().loginSuccessDto = responseVo.content;
                    System.out.println("session更新成功");

                    //如果失败了不再自动登录，直接返回对应的请求处理对象
                    return getResponseVoObservable(api);
                    //一直到登录成功为止
                    //return login(api);
                } else {
                    return Observable.create(new ObservableOnSubscribe<T>() {
                        @Override
                        public void subscribe(ObservableEmitter<T> observableEmitter) throws Exception {
                            System.err.println(responseVo);
                            observableEmitter.onError(new MyException(responseVo.errmsg,
                                    responseVo.errcode));
                        }
                    });
                }
            }
        });
    }
    // 得到期待的 Observable<HttpBody>
    public <T> Observable<T> getResponseVoObservable(Observable<ResponseVo<T>> observable) {
        return observable.concatMap(new Function<ResponseVo<T>, ObservableSource<T>>() {
            @Override
            public ObservableSource<T> apply(final ResponseVo<T> responseVo) throws Exception {
                //System.out.println(integer);
                return valedateCode(responseVo);
            }
        });
    }

    //errcode=0 表示请求成功了
    private <T> Observable<T> valedateCode(final ResponseVo<T> responseVo) {
        return Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(ObservableEmitter<T> observableEmitter) throws Exception {
                if (responseVo.errcode != null && responseVo.errcode.equals("0")) {//成功
                    if (responseVo.content != null)
                        observableEmitter.onNext(responseVo.content);
                    observableEmitter.onComplete();
                } else {
                    System.err.println("valedateCode:" + responseVo);
                    observableEmitter.onError(new MyException(responseVo.errmsg, responseVo.errcode));
                }
            }
        });
    }
}