RXjava业务常用功能

引用的是Rxjava3.0版本
1、按钮倒计时

 /**
     * 按钮倒计时
     */
    private fun initbtnTimer() {
        disposable = Flowable.intervalRange(0, 15, 0, 1, TimeUnit.SECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext(Consumer { time ->
                mBinding?.okTv?.apply {
                    setText((15 - time).toString())
                    isClickable = false
                }
            })
            .doOnComplete({
                mBinding?.okTv?.apply {
                    setText("我同意")
                    isClickable = true
                    requestFocus()
                    background =
                        ContextCompat.getDrawable(context, R.drawable.tv_bg_normal_text_corner_4)
                }
            })
            .subscribe()
    }

    override fun dismiss() {
        super.dismiss()
        disposable?.dispose()

    }

2、延迟执行异步任务然后更新UI

 Observable.timer(500, TimeUnit.MILLISECONDS)
            .doOnNext({
                Log.d("TAG", "thred " + Thread.currentThread().name)
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                Log.d("TAG", "subscribe thred " + Thread.currentThread().name)
                storageProgressBar.calcPosBySize(0f, 0.2f, 0.3f)
            })

3、先注册后登录

   /**
     * 获取当前时间戳
     *
     * @return
     */
    public void login() {
        String account = "786928736@qq.com";
        String country = "86";
        progressBarLiveData.setValue(true);
        BrainService.getService()
                .getBaseUrlByEmail(account, country)
                .compose(RxUtils.exceptionTransformer()) // 网络错误的异常转换, 这里可以换成自己的ExceptionHandle
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<BaseResponse<BaseUrlResult>>() {
                    @Override
                    public void accept(BaseResponse<BaseUrlResult> response) throws Exception {
                        if (response != null && response.getCode() == StateCode.SUCCESS.code) {
                            BaseUrlResult result = response.getData();
                            PrefsUtil.putString(TeckinApplication.mCtx, PrefsUtil.GLOBAL_PREFS_FILE_NAME, PrefsUtil.KEY_WEB_URL, result.getWeb());
                            PrefsUtil.putString(TeckinApplication.mCtx, PrefsUtil.GLOBAL_PREFS_FILE_NAME, PrefsUtil.KEY_S3_URL, result.getS3());
                            PrefsUtil.putString(TeckinApplication.mCtx, PrefsUtil.GLOBAL_PREFS_FILE_NAME, PrefsUtil.KEY_REGION, result.getRegion());
                            PrefsUtil.putString(TeckinApplication.mCtx, PrefsUtil.GLOBAL_PREFS_FILE_NAME, PrefsUtil.KEY_P2P_URL, result.getP2p());
                        }
                    }
                }).observeOn(Schedulers.io())
                .flatMap(new Function<BaseResponse<BaseUrlResult>, ObservableSource<BaseResponse<BaseUrlResult>>>() {
                    @Override
                    public ObservableSource<BaseResponse<BaseUrlResult>> apply(@io.reactivex.annotations.NonNull BaseResponse<BaseUrlResult> response) throws Exception {
                        return BrainService.getService().getBaseUrlByAccount(account);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<BaseResponse<BaseUrlResult>>() {
                    @Override
                    public void onSubscribe(@io.reactivex.annotations.NonNull Disposable d) {
                        Disposable disposable = d;
                    }

                    @Override
                    public void onNext(@io.reactivex.annotations.NonNull BaseResponse<BaseUrlResult> response) {
                        int code = response.getCode();
                    }

                    @Override
                    public void onError(@io.reactivex.annotations.NonNull Throwable e) {
                        e.printStackTrace();

                        if (e instanceof ResponseThrowable) {
                            Toast.makeText(TeckinApplication.mCtx, ((ResponseThrowable) e).message, Toast.LENGTH_LONG).show();
                        }
                    }

                    @Override
                    public void onComplete() {
                        progressBarLiveData.setValue(false);
                        int i = 100;
                    }
                });

        //
    }

zip操作
比如一个界面需要展示用户的一些信息, 而这些信息分别要从两个服务器接口中获取, 而只有当两个都获取到了之后才能进行展示, 这个时候就可以用Zip了

/**
     *
     */
    fun requestTempleAndMediaInfo(orderId: String) {
        Observable.zip(
            requestTempletaDatasObservable(orderId),
            requestMediaInfoObservable(orderId),
            object : BiFunction<MediaTemplateModel, MediaInfoModel, MediaTemplateModel> {
                override fun apply(
                    templateModel: MediaTemplateModel,
                    mediaInfo: MediaInfoModel
                ): MediaTemplateModel {
                    mediaInfoJoinTemple(templateModel, mediaInfo)
                    return templateModel
                }

            }
        ).observeOn(Schedulers.io())
            .subscribe(object : io.reactivex.Observer<MediaTemplateModel> {
                override fun onSubscribe(p0: Disposable) {

                }

                override fun onNext(media: MediaTemplateModel) {
                    dealDataGetShowData(media)
                }

                override fun onError(p0: Throwable) {

                }

                override fun onComplete() {
                }

            })

    }