/**
 * fetch leaderboard
 */
private void loadLeaders() {
  RetrofitServiceGenerator.getService().getLeaderboardList()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeWith(new SingleObserver<LeaderboardModel>() {
      @Override
      public void onSubscribe(Disposable d) {
        compositeDisposable.add(d);
      }

      @Override
      public void onSuccess(LeaderboardModel leaderboardModel) {
        handleLeaderBoardList(leaderboardModel);
      }

      @Override
      public void onError(Throwable e) {
      }
    });
}
 

/**
 * 初始化提醒时间
 */
private void initDuraion() {
    io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            SharedPreferences sharedPreferences =
                    getSharedPreferences("com.android.betterway_preferences", Context.MODE_PRIVATE);
            weatherWarn = sharedPreferences.getBoolean("weather_warn", false);
            if (weatherWarn) {
                String d = sharedPreferences.getString("warning_time", "15");
                int dur = Integer.parseInt(d);
                e.onNext(dur);
            } else {
                e.onNext(0);
            }
        }
    }).observeOn(Schedulers.io())
            .subscribeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    duration = integer;
                }
            });
}
 

@SuppressLint("CheckResult")
@Override
public void load(final MutableLiveData<String> result) {
    Observable
            .fromCallable(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return "DataLoadDemo11 called";
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    result.setValue(s);
                }
            });
}
 

/**
 * 从设置页面获取用户信息并返回结果
 *
 * @param responseListener 请求结果回调接口
 */
public static void getInfo(ResponseListener<Account> responseListener){

    mUserApi.getSettingPage()
            .compose(RxUtil.io2computation())
            .flatMap((Function<String, ObservableSource<Account>>) s -> {
                ErrorEnum.ERR_PAGE_NEED_LOGIN.check(s);
                ErrorEnum.ERR_PAGE_NEED_LOGIN0.check(s);
                return getAccount(s);
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new RxObserver<Account>() {
                @Override
                public void _onError(String msg) {
                    super._onError(msg);
                    responseListener.onFailed(msg);
                }
                @Override
                public void _onNext(Account account) {
                    responseListener.onComplete(account);
                }
            });
}
 

public void requestDetailInfo(int id){
    mModel.getDetailInfo(id)
            .subscribeOn(Schedulers.io())
            .retryWhen(new RetryWithDelay(3, 2))//遇到错误时重试,第一个参数为重试几次,第二个参数为重试的间隔
            .doOnSubscribe(disposable -> {
                mRootView.showLoading();//显示下拉刷新的进度条
            }).subscribeOn(AndroidSchedulers.mainThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doFinally(() -> {
                mRootView.hideLoading();//隐藏下拉刷新的进度条
            })
            .compose(RxLifecycleUtils.bindToLifecycle(mRootView))//使用 Rxlifecycle,使 Disposable 和 Activity 一起销毁
            .subscribe(new ErrorHandleSubscriber<ZhihuDetailBean>(mErrorHandler) {
                @Override
                public void onNext(ZhihuDetailBean zhihuDetailBean) {
                    mRootView.shonContent(zhihuDetailBean);
                }
            });
}
 

@Override
protected void initClick() {
    super.initClick();
    mBookListAdapter.setOnLoadMoreListener(
            () -> {
               mPresenter.loadBookList(mBookListType,mTag,mStart,mLimit);
            }
    );
    mBookListAdapter.setOnItemClickListener(
            (view,pos) -> {
                BookListBean bean = mBookListAdapter.getItem(pos);
                BookListDetailActivity.startActivity(getContext(),bean.get_id());
            }
    );

    Disposable disposable = RxBus.getInstance()
            .toObservable(BookSubSortEvent.class)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    event -> {
                        mTag = event.bookSubSort;
                        showRefresh();
                    }
            );
    addDisposable(disposable);
}
 

private void wireData() {
  subscriptions = new CompositeDisposable();
  subscriptions.add(Select
      .from(ITEM)
      .where(ITEM.LIST.is(itemList))
      .orderBy(ITEM.COMPLETE.asc(), ITEM.DESCRIPTION.asc())
      .observe()
      .runQuery()
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(adapter::setData));
  subscriptions.add(Observable.combineLatest(Select
          .from(ITEM_LIST)
          .where(ITEM_LIST.ID.is(itemList.id()))
          .takeFirst()
          .observe()
          .runQuery(),
      Item.countItemsFor(itemList)
          .runQuery(),
      (itemList, count) -> String.format("%s (%s)", itemList.name(), count))
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(this::setTitle));
}
 

public void onSeed(String walletAddress, KeyService.AuthenticationLevel level)
{
    if (walletAddress == null)
    {
        progress.postValue(false);
        System.out.println("ERROR");
        badSeed.postValue(true);
    }
    else
    {
        progress.postValue(true);
        //begin key storage process
        disposable = importWalletInteract.storeHDWallet(walletAddress, level, ensResolver)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(wallet::postValue, this::onError); //signal to UI wallet import complete
    }
}
 

public void loadAnnouncementDetails(String id) {
    checkViewAttached();
    compositeDisposable.add(mDataManager.getAnnouncementDetail(id)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribeWith(new DisposableObserver<DetailAnnouncement>() {

                @Override
                public void onNext(DetailAnnouncement detailAnnouncement) {
                    getMvpView().showAnnouncementDetail(detailAnnouncement);
                }

                @Override
                public void onError(Throwable e) {
                    Log.d(LOG_TAG, e.getMessage());
                    getMvpView().showWaitProgress(false);
                    getMvpView().showSnackBar(R.string.failed_to_fetch_announcement);
                }

                @Override
                public void onComplete() {
                    getMvpView().showWaitProgress(false);
                }
            }));
}
 

private void query_variables_directives() {
    okGraphql

            .query(
                    "Hero($episode: Episode, $withFriends: Boolean!) {" +
                            "  hero(episode: $episode) {" +
                            "    name" +
                            "    friends @include(if: $withFriends) {" +
                            "      name" +
                            "    }" +
                            "  }"
            )


            .variable("episode", "JEDI")
            .variable("withFriends", false)

            .cast(StarWarsResponse.class)

            .toSingle()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    data -> text1.setText(data.toString()),
                    throwable -> text1.setText(throwable.getLocalizedMessage())
            );
}
 

public Disposable getWhatsNewIcons() {
    return Observable.fromArray(mView.getResources().getStringArray(R.array.whatsnew))
            .map(new Function<String, IconBean>() {
                @Override
                public IconBean apply(@NonNull String s) throws Exception {
                    IconBean bean = new IconBean();
                    bean.id = mView.getResources().getIdentifier(s, "drawable", BuildConfig.APPLICATION_ID);
                    bean.name = s;

                    return bean;
                }
            }).toList().subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<List<IconBean>>() {
                @Override
                public void accept(List<IconBean> list) throws Exception {
                    mView.onLoadData(list);
                }
            });
}
 

@Override
protected void onResume() {
    super.onResume();
    presenter.onResume();
    navigatorHolder.setNavigator(navigator);

    MessagesPresenter localPresenter = presenter;
    messageInputDisposable = RxTextView
            .textChanges(newMessageText)
            .debounce(500, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    localPresenter::onChangeMessageText,
                    error -> LoggerHelper.e(getClass().getSimpleName(), error.getMessage(), error)
            );

    endlessScrollListener = new EndlessUpScrollListener(layoutManager) {
        @Override
        public void onLoadMore(int page, int totalItemsCount, RecyclerView view) {
            presenter.loadMore();
        }
    };
    endlessScrollListener.setVisibleThreshold(12);
    messagesList.addOnScrollListener(endlessScrollListener);
}
 

private void initSearchFeature(StudentAdapter adapter) {
    RxSearchObservable.fromSearchView(svKey)
            .debounce(500, TimeUnit.MILLISECONDS)
            .filter(text -> !text.isEmpty())
            .distinctUntilChanged()
            .switchMap(new Function<String, ObservableSource<ArrayList<String>>>() {
                @Override
                public ObservableSource<ArrayList<String>> apply(@NonNull String key) throws Exception {
                    return DataSource.getSearchData(key);
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(names -> {
                adapter.removeAllNames();
                adapter.addStudentNames(names);
            });
}
 

@Override
public void play(Activity activity) {
    // 播放
    if (TextUtils.isEmpty(mSelMp3Path)) {
        Toast.makeText(activity,
                activity.getString(R.string.dialog_cutter_warning_sel),
                Toast.LENGTH_SHORT).show();
        return;
    }
    mMediaPlayer.start();
    mView.setPlayBtnWithStatus(true);
    mView.setVisualizerViewEnaled(true);
    seekToForIsMin();
    //开启进度rx轮询事件
    mDisposable = mUpdateProgressObservable.observeOn(AndroidSchedulers.mainThread()).
            subscribe(mUpdateProgressConsumer);
}
 

public Single<Wallet> find() {
	return walletRepository
			.getDefaultWallet()
			.onErrorResumeNext(walletRepository
					.fetchWallets()
					.to(single -> Flowable.fromArray(single.blockingGet()))
					.firstOrError()
					.flatMapCompletable(walletRepository::setDefaultWallet)
					.andThen(walletRepository.getDefaultWallet()))
			.observeOn(AndroidSchedulers.mainThread());
}
 

public void  searchBook(String bookName,int page,Observer<List<SearchBook>> sub) {
    ApiUtils.getApi(AppConfig.idsPwd).create(OpacApi.class).searchBook(bookName,page)
            .map(new HttpResultFunc<List<SearchBook>>())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(sub);
}
 

protected Disposable startApiAsyncTask(String apiName, FunctionWithApiResponse function) {
    progressDialog.setCancelable(false);
    progressDialog.show();

    return Single.just(apiName)
            .subscribeOn(Schedulers.io())
            .map(name -> function.method())
            .observeOn(AndroidSchedulers.mainThread())
            .doFinally(() -> progressDialog.dismiss())
            .subscribe(
                    lineApiResponse -> addLog("\n== " + apiName + " == " + lineApiResponse.getResponseCode() + LOG_SEPARATOR + lineApiResponse),
                    error -> addLog("\n== " + apiName + " == Error\n" + error.getMessage())
            );
}
 

@SuppressLint("CheckResult")
@Override
public void fetchUserInfo(final BaseToken token) {
    mSubscribe = Flowable.create(new FlowableOnSubscribe<WeiboUser>() {
        @Override
        public void subscribe(@NonNull FlowableEmitter<WeiboUser> weiboUserEmitter) {
            OkHttpClient client = new OkHttpClient();
            Request request = new Request.Builder().url(buildUserInfoUrl(token, USER_INFO)).build();
            try {
                Response response = client.newCall(request).execute();
                JSONObject jsonObject = new JSONObject(response.body().string());
                WeiboUser user = WeiboUser.parse(jsonObject);
                weiboUserEmitter.onNext(user);
                weiboUserEmitter.onComplete();
            } catch (IOException | JSONException e) {
                ShareLogger.e(ShareLogger.INFO.FETCH_USER_INOF_ERROR);
                weiboUserEmitter.onError(e);
            }
        }
    }, BackpressureStrategy.DROP)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<WeiboUser>() {
                @Override
                public void accept(@NonNull WeiboUser weiboUser) throws Exception {
                    mLoginListener.loginSuccess(
                            new LoginResultData(LoginPlatform.WEIBO, token, weiboUser));
                    LoginUtil.recycle();
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(@NonNull Throwable throwable) throws Exception {
                    mLoginListener.loginFailure(new Exception(throwable), ShareLogger.INFO.ERR_FETCH_CODE);
                    LoginUtil.recycle();
                }
            });
}
 

public static <T> ObservableTransformer<T, T> ioMainThread() {
    return new ObservableTransformer<T, T>() {
        @Override
        public ObservableSource<T> apply(Observable<T> upstream) {
            return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    };
}
 

private void getActiveEnrollments() {
    compositeDisposable.add(teiProgramListRepository.activeEnrollments(trackedEntityId)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(enrollments -> {
                        Collections.sort(enrollments, (enrollment1, enrollment2) -> enrollment1.programName().compareToIgnoreCase(enrollment2.programName()));
                        view.setActiveEnrollments(enrollments);
                    },
                    Timber::d)
    );
}
 

private void listen() {
    stopListen();
    if (mDisposable == null) {
        mDisposable = RxBus.listen(PortraitEvent.class)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<PortraitEvent>() {
                @Override
                public void accept(PortraitEvent portraitEvent) throws Exception {
                    if (portraitEvent.getKey().equals(mFriendKey)) {
                        PortraitView.this.loadAvatar();
                    }
                }
            });
    }
}
 

void restoreOps(final RestoreModel model) {
  final int size = model.preAppInfos.size();
  final AtomicInteger progress = new AtomicInteger();
  mView.showProgress(true, size);

  Observable.fromIterable(model.preAppInfos)
      .flatMap(new Function<PreAppInfo, ObservableSource<OpsResult>>() {
        @Override
        public ObservableSource<OpsResult> apply(@NonNull PreAppInfo appInfo) throws Exception {
          return Helper.setModes(context, appInfo.getPackageName(), AppOpsManager.MODE_IGNORED,
              appInfo.getOps());
        }
      }).subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread()).subscribe(new ResourceObserver<OpsResult>() {
    @Override
    public void onNext(@NonNull OpsResult opsResult) {
      mView.setProgress(progress.incrementAndGet());
    }

    @Override
    public void onError(@NonNull Throwable e) {
      progress.incrementAndGet();
    }

    @Override
    public void onComplete() {
      mView.showProgress(false, 0);
      Toast.makeText(context, "恢复成功", Toast.LENGTH_LONG).show();
    }
  });
}
 

private void requestUsers() {
    Disposable subscription = usersService.getUsers(1)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .retry(3)
            .subscribe(userModels -> usersAdapter.addData(userModels),
                    Throwable::printStackTrace);
    compositeDisposable.add(subscription);
}
 

private void loadMovieDetail() {
    HttpClient.Builder.getDouBanService().getMovieDetail(subjectsBean.getId())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<MovieDetailBean>() {

                @Override
                public void onError(Throwable e) {
                    showError();
                }

                @Override
                public void onComplete() {
                    showContentView();
                }

                @Override
                public void onSubscribe(Disposable d) {
                    addSubscription(d);
                }

                @Override
                public void onNext(final MovieDetailBean movieDetailBean) {
                    // 上映日期
                    bindingHeaderView.tvOneDay.setText(String.format("上映日期：%s", movieDetailBean.getYear()));
                    // 制片国家
                    bindingHeaderView.tvOneCity.setText(String.format("制片国家/地区：%s", StringFormatUtil.formatGenres(movieDetailBean.getCountries())));
                    bindingHeaderView.setMovieDetailBean(movieDetailBean);
                    bindingContentView.setBean(movieDetailBean);
                    bindingContentView.executePendingBindings();

                    mMoreUrl = movieDetailBean.getAlt();
                    mMovieName = movieDetailBean.getTitle();

                    transformData(movieDetailBean);
                }
            });
}
 

public static void follow(String animaId, CommJsonObserver<CommJsonEntity> observer, NetworkConsumer consumer){
    Map<String, String> map = new HashMap<>();
    map.put("animeId", animaId);
    map.put("favoriteStatus", "favorited");
    map.put("rating", "0");
    RetroFactory.getInstance().follow(map)
            .doOnSubscribe(consumer)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer);
}
 

private void getENSAddress(String name)
{
    ensProgress();
    if (ensResolver == null)
        ensResolver = new AWEnsResolver(TokenRepository.getWeb3jService(EthereumNetworkRepository.MAINNET_ID), getContext());
    ensResolver.resolveENSAddress(name)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::fetchedENSAddress, this::onENSFail).isDisposed();
}
 

private void updateYoutubeDL() {
    if (updating) {
        Toast.makeText(MainActivity.this, "update is already in progress", Toast.LENGTH_LONG).show();
        return;
    }

    updating = true;
    progressBar.setVisibility(View.VISIBLE);
    Disposable disposable = Observable.fromCallable(() -> YoutubeDL.getInstance().updateYoutubeDL(getApplication()))
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(status -> {
                progressBar.setVisibility(View.GONE);
                switch (status) {
                    case DONE:
                        Toast.makeText(MainActivity.this, "update successful", Toast.LENGTH_LONG).show();
                        break;
                    case ALREADY_UP_TO_DATE:
                        Toast.makeText(MainActivity.this, "already up to date", Toast.LENGTH_LONG).show();
                        break;
                    default:
                        Toast.makeText(MainActivity.this, status.toString(), Toast.LENGTH_LONG).show();
                        break;
                }
                updating = false;
            }, e -> {
                if(BuildConfig.DEBUG) Log.e(TAG, "failed to update", e);
                progressBar.setVisibility(View.GONE);
                Toast.makeText(MainActivity.this, "update failed", Toast.LENGTH_LONG).show();
                updating = false;
            });
    compositeDisposable.add(disposable);
}
 

public void remove(@NonNull Content content) {
    compositeDisposable.add(
            Completable.fromRunnable(() -> doRemove(content.getId()))
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(
                            () -> {
                                // Nothing to do here; UI callbacks are handled through LiveData
                            },
                            Timber::e
                    )
    );
}
 

public void enableTwoFactor(final String method, final String data) {
    if (twoFactorConfigData.getEnabledMethods().contains(method.toLowerCase(Locale.US))) {
        UI.toast(this, "Two factor just enabled", Toast.LENGTH_LONG);
        finishOnUiThread();
        return;
    }

    disposable = Observable.just(getSession())
                 .subscribeOn(Schedulers.computation())
                 .map((session) -> {
        final TwoFactorDetailData twoFactorDetail = new TwoFactorDetailData();
        twoFactorDetail.setEnabled(true);
        twoFactorDetail.setData(data);
        twoFactorDetail.setConfirmed(true);
        session.changeSettingsTwoFactor(method, twoFactorDetail).resolve(new PopupMethodResolver(this),
                                                                         new PopupCodeResolver(this));
        return session;
    })
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe((session) -> {
        UI.toast(this, "Two factor enabled", Toast.LENGTH_LONG);
        setEnableDisableResult(true);
        finishOnUiThread();
    }, (final Throwable e) -> {
        UI.toast(this, e.getMessage(), Toast.LENGTH_LONG);
    });
}
 

public void search(final String key) {
    bookChapterBeans.clear();
    if (Objects.equals(key, "")) {
        isSearch = false;
        notifyDataSetChanged();
    } else {
        Observable.create((ObservableOnSubscribe<Boolean>) emitter -> {
            for (BookChapterBean bookChapterBean : allChapter) {
                if (bookChapterBean.getDurChapterName().contains(key)) {
                    bookChapterBeans.add(bookChapterBean);
                }
            }
            emitter.onNext(true);
            emitter.onComplete();
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new MyObserver<Boolean>() {
                    @Override
                    public void onNext(Boolean aBoolean) {
                        isSearch = true;
                        notifyDataSetChanged();
                    }

                    @Override
                    public void onError(Throwable e) {

                    }
                });
    }
}