2021-05-28 Rxjava concatMap按顺序

最近在项目想出的解决方法，断网时候，记录操作，把数据保存到list里面，等网络来了后，按顺序取出list的内容，然后依次请求网络。

1.定义存放操作的list,里面的map是每次操作请求的参数，在这里添加了两个参数，用来区别这个map是执行的什么操作。

    /**
     * 存放离线时的用户操作
     * map为用户请求的参数，用key值"operation"区分操作
     * 添加 "add" , 删除 "delete"，编辑 "edit"
     */
    public static List<Map> noNetWorkEditAddrsList = new ArrayList<Map>();

2.重写BaseObserver，添加断网时的操作方法，这样继承它的方法都有了断网操作方法

public abstract class BaseObserver<T> implements Observer<T> {

    private static final String TAG = "BaseSubscriber";
    private Context mContext;

    public BaseObserver(Context context) {
        this.mContext = context;
    }

    @Override
    public void onSubscribe(@NonNull Disposable d) {
        if (!NetworkUtil.isNetworkConnected(mContext)) {
            Log.d(TAG, "onSubscribe: 当前网络不可用，请检查网络情况");
            onNoNetWork();
            onComplete();
            return;
        }
        Log.d(TAG, "onSubscribe: network oK!");
    }

    public void onNoNetWork(){

    }

3.在接口的回调方法内记录断网时的操作。（将请求map放入list内）

                    @Override
                    public void onNoNetWork() {
                        super.onNoNetWork();
                        Log.d(TAG, "onNoNetWork: setAddAddrCollectToTSP , waiting for net");
                        requestAddrMap.put("operation","add");
                        noNetWorkEditAddrsList.add(requestAddrMap);
                    }

4.收到有网的广播后，使用concatMap顺序请求接口，优点是每次请求完成，才会开始下一次请求，对于参数需要更新的嵌套请求非常管用

**
     * 网络来了后，依次执行离线的操作
     */
    public void syncNoNetWorkOperation(){
        Observable.fromIterable(noNetWorkEditAddrsList)
                .concatMap(new Function<Map, ObservableSource<BaseResponse<CollectAddressList>>>(){
                    @Override
                    public ObservableSource<BaseResponse<CollectAddressList>> apply(@NonNull Map map) throws Exception {
                        Log.d(TAG, "apply: noNetWorkEditAddrsList.size=" + noNetWorkEditAddrsList.size());
                        Observable<BaseResponse<CollectAddressList>> observable = null;

                        String operation = map.get("operation").toString();
                        switch (operation) {
                            case "add":
                                Log.d(TAG, "apply: add");
                                map.remove("operation");
                                observable = RetrofitUtil.getInstance().getRetrofitInterface().addAddrCollect(map);
                                return observable;
                            case "delete":
                                map.remove("operation");
                                //删除地址，需要传入服务器返回的addrId,
                                int type = (int) map.get("type");
                                String addrId = (type == FAVORITE_TYPE_HOME) ?LOCAL_HOME_ADDRS_ID:LOCAL_COMPANY_ADDRS_ID;
                                Log.d(TAG, "apply delete : addrId = "+addrId + "type = "+ type);
                                map.put(REQUEST_ADDR_ID, addrId);
                                observable = RetrofitUtil.getInstance().getRetrofitInterface().delAddrCollect(map);
                                return observable;
                            case "edit":
                                map.remove("operation");
                                String addrName = map.get("addrName").toString();
                                String id = DBHelper.getInstance(mContext).queryDBByName(addrName);
                                Log.d(TAG, "apply edit: id = "+id +"addrName = "+addrName);
                                map.put(REQUEST_ID, id);
                                observable = RetrofitUtil.getInstance().getRetrofitInterface().editAddAddrCollect(map);
                                return observable;
                            default:
                                Log.d(TAG, "apply: default error");
                                map.remove("operation");
                                observable = RetrofitUtil.getInstance().getRetrofitInterface().addAddrCollect(map);
                                return observable;
                        }
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(new BaseObserver<BaseResponse<CollectAddressList>>(mContext) {
                    @Override
                    public void onNext(@NonNull BaseResponse<CollectAddressList> baseResponse) {
                        if (baseResponse.isOk(mContext)) {
                            if(baseResponse.getData() != null){
                                Log.d(TAG, "onNext: haveNetWorkEditAddrs SUCCESS");
                                UpdateInfo.updateAddressInfo(mContext, baseResponse.getData());
                            }
                        } else {
                            Log.d(TAG, "onNext: haveNetWorkEditAddrs FAILURE");
                        }
                    }

                    @Override
                    public void onComplete() {
                        super.onComplete();
                        Log.d(TAG, "onComplete:");
                        //list请求完后执行：请求服务器数据，将TSP改动同步到本地数据库
                        noNetWorkEditAddrsList.clear();
                        getUserInfoFromTSP();
                        getCollectAddressListFromTSP();
                    }
                });

5.执行完后，不要忘了在onComplete 里面清空list，这样下次断网联网不会请求上次保存的操作.

noNetWorkEditAddrsList.clear();