Java 类rx.schedulers.Schedulers 实例源码

private void scanBleDevices() {
    mScanSubscroption = mRxBleClient
            .scanBleDevices()
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .filter(rxBleScanResult -> {
                return FilterDeviceName.equals(rxBleScanResult.getBleDevice().getName());
            })
            .subscribe(rxBleScanResult -> {
                int position = DeviceContext.AddorUpdateDevice(rxBleScanResult.getBleDevice()
                                .getMacAddress(),
                        rxBleScanResult.getRssi());
                if (position >= 0)
                    mScanDevices.getAdapter().notifyItemChanged(position);
                else
                    mScanDevices.getAdapter().notifyDataSetChanged();
            }, this::onScanFailure);
}

@Override
public void onActivityCreated(@Nullable Bundle savedInstanceState) {
    super.onActivityCreated(savedInstanceState);
    mAccount = getArguments().getParcelable("account");
    mUserListsModel = new UserListsModel(mAccount);

    mSubscriptions.add(
            mUserListsModel
                    .lists()
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                    .subscribe(new DefaultObserver<List<UserList>>() {
                        @Override
                        public void onNext(List<UserList> userLists) {
                            super.onNext(userLists);
                            mRecyclerView.setAdapter(new ListAdapter(getActivity(), mAccount, userLists));
                        }
                    })
    );
}

@Override
public void loadData()
{

    RetrofitHelper.getLiveAPI()
            .getLiveAppIndex()
            .compose(bindToLifecycle())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(liveAppIndexInfo -> {

                mLiveAppIndexAdapter.setLiveInfo(liveAppIndexInfo);
                finishTask();
            }, throwable -> {
                initEmptyView();
            });
}

public static RetrofitInterface getRetrofit(String email, String password) {

        String credentials = email + ":" + password;
        String basic = "Basic " + Base64.encodeToString(credentials.getBytes(), Base64.NO_WRAP);
        OkHttpClient.Builder httpClient = new OkHttpClient.Builder();

        httpClient.addInterceptor(chain -> {

            Request original = chain.request();
            Request.Builder builder = original.newBuilder()
                    .addHeader("Authorization", basic)
                    .method(original.method(),original.body());
            return  chain.proceed(builder.build());

        });

        RxJavaCallAdapterFactory rxAdapter = RxJavaCallAdapterFactory.createWithScheduler(Schedulers.io());

        return new Retrofit.Builder()
                .baseUrl(Constants.BASE_URL)
                .client(httpClient.build())
                .addCallAdapterFactory(rxAdapter)
                .addConverterFactory(GsonConverterFactory.create())
                .build().create(RetrofitInterface.class);
    }

/**
 * 图片开始轮播
 */
private void startScroll()
{

    compositeSubscription = new CompositeSubscription();
    isStopScroll = false;
    Subscription subscription = Observable.timer(delayTime, TimeUnit.SECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(aLong -> {

                if (isStopScroll)
                    return;

                isStopScroll = true;
                viewPager.setCurrentItem(viewPager.getCurrentItem() + 1);
            });
    compositeSubscription.add(subscription);
}

public void getUserInfo() {

    RetrofitHelper.getAccountAPI()
        .getUserInfoById(mid)
        .compose(this.bindToLifecycle())
        .doOnSubscribe(this::showProgressBar)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(userInfo -> {

          mUserDetailsInfo = userInfo;
          finishTask();
        }, throwable -> {
          hideProgressBar();
        });
  }

public void doSubscribe() {
    subscription = Observable.just("Hello Reactive!").observeOn(AndroidSchedulers.mainThread()).subscribeOn(Schedulers.io()).subscribe(new Subscriber<String>() {
        @Override
        public void onCompleted() {
            Toast.makeText(MainActivity.this, "Finish Subscribing...", Toast.LENGTH_SHORT).show();
        }

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

        @Override
        public void onNext(final String s) {
            Toast.makeText(MainActivity.this, s, Toast.LENGTH_SHORT).show();
        }
    });
}

private void initializeData() {

        application.getWebService()
                .getInteractionPosts(likeCounter, MySharedPreferences.getUserId(preferences), MySharedPreferences.getUserToken(preferences))
                .retryWhen(new RetryWithDelay(3,2000))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new BaseSubscriber<PostSuperUserListModel>() {
                    @Override
                    public void onNext(PostSuperUserListModel response) {
                        showRecycleWithDataFilled(response);
                        progressFrame.setVisibility(View.GONE);
                    }
                    @Override
                    public void onError(Throwable e) {
                        progressFrame.setVisibility(View.GONE);
                        try {
                            Timber.e(e.getMessage());
                     //       Toast.makeText(UserSameCounterActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
                        }catch (Exception ex){
                            ex.printStackTrace();
                        }
                    }
                });
    }

@Override
public void getBookListDetail(String bookListId) {
    Subscription rxSubscription = bookApi.getBookListDetail(bookListId).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<BookListDetail>() {
                @Override
                public void onCompleted() {
                    mView.complete();
                }

                @Override
                public void onError(Throwable e) {
                    LogUtils.e("getBookListDetail:" + e.toString());
                    mView.complete();
                }

                @Override
                public void onNext(BookListDetail data) {
                    mView.showBookListDetail(data);
                }
            });
    addSubscrebe(rxSubscription);
}

@Override
public void getBestComments(String disscussionId) {
    Subscription rxSubscription = bookApi.getBestComments(disscussionId)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<CommentList>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    LogUtils.e("getBestComments:" + e.toString());
                }

                @Override
                public void onNext(CommentList list) {
                    mView.showBestComments(list);
                }
            });
    addSubscrebe(rxSubscription);
}

public void getOrders(int page,int pageNum,String startTime,String endTime){
    checkViewAttached();
    RxUtil.unsubscribe(mGetOrdersSubscription);
    mGetOrdersSubscription = mDataManager.getOrderList(page,pageNum,startTime,endTime)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<OrderResponse>() {
                @Override
                public void onCompleted() {
                }

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

                @Override
                public void onNext(OrderResponse orderResponse) {
                    getMvpView().showOrders(orderResponse);
                }
            });
}

private void changeMyName() {
    showWaitingDialog(UIUtils.getString(R.string.please_wait));
    String nickName = mEtName.getText().toString().trim();
    ApiRetrofit.getInstance().setName(nickName)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(setNameResponse -> {
                hideWaitingDialog();
                if (setNameResponse.getCode() == 200) {
                    Friend friend = DBManager.getInstance().getFriendById(UserCache.getId());
                    if (friend != null) {
                        friend.setName(nickName);
                        friend.setDisplayName(nickName);
                        DBManager.getInstance().saveOrUpdateFriend(friend);
                        BroadcastManager.getInstance(ChangeMyNameActivity.this).sendBroadcast(AppConst.CHANGE_INFO_FOR_ME);
                        BroadcastManager.getInstance(ChangeMyNameActivity.this).sendBroadcast(AppConst.CHANGE_INFO_FOR_CHANGE_NAME);
                    }
                    finish();
                }
            }, this::loadError);
}

protected void sendLikeNotification(String likeUrl) {
    application.getWebService()
            .sendLikeNotification(likeUrl)
            .retryWhen(new RetryWithDelay(3,2000))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new BaseSubscriber<String>() {
                @Override
                public void onNext(String response) {
                    Timber.d("Got user details");
                    //     followers.setText(String.valueOf(response.size()));
                    // Toast.makeText(ChangeProfileActivity.this, "Message Sent", Toast.LENGTH_SHORT).show();
                    //  Timber.d("Message from server" + response);
                }
                @Override
                public void onError(Throwable e) {
                    Crashlytics.logException(e);
                    try {
                        Toast.makeText(PostsDetailsActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
                    }catch (Exception ex){
                        ex.printStackTrace();
                    }
                }
            });
}

@Override
protected void loadData() {
    RetrofitHelper.getBiliAppAPI()
            .getRecommendedBannerInfo()
            .compose(bindToLifecycle())
            .map(RecommendBannerInfo::getData)
            .flatMap(new Func1<List<RecommendBannerInfo.DataBean>, Observable<RecommendInfo>>() {
                @Override
                public Observable<RecommendInfo> call(List<RecommendBannerInfo.DataBean> dataBeans) {
                    mRecommendedBanners.addAll(dataBeans);
                    return RetrofitHelper.getBiliAppAPI().getRecommendedInfo();
                }
            })
            .compose(bindToLifecycle())
            .map(RecommendInfo::getResult)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(resultBeans -> {
                mResults.addAll(resultBeans);
                finishTask();
            }, throwable -> {
                initEmptyView();
            });
}

public void dealApiReqest(BaseApi baseApi) {
//                创建一个OkHttpClient
                OkHttpClient.Builder builder = new OkHttpClient.Builder();
                CommonSubscriber commonSubscriber = new CommonSubscriber(baseApi, mOnNextListenerSoftReference, mRxAppCompatActivitySoftReference);
                builder.addInterceptor(new CacheInterceptor()).connectTimeout(baseApi.getConnectedTime(), TimeUnit.SECONDS)
                        .writeTimeout(baseApi.getWritedOutTime(), TimeUnit.SECONDS).addNetworkInterceptor(new CacheInterceptor())
                        .cache(new Cache(CustomApplication.getInstance().getCacheDir(), 10 * 1024 * 1024));
                Retrofit retrofit = new Retrofit.Builder().client(builder.build()).baseUrl(baseApi.getBaseUrl()).addConverterFactory(ScalarsConverterFactory.create())
                        .addCallAdapterFactory(RxJavaCallAdapterFactory.create()).build();
//                        .build();
//                这里原理是通过Java的动态代理创建接口的实例
                HttpService httpService = retrofit.create(HttpService.class);
                Observable observable = baseApi.getObservable(httpService);
                observable.retryWhen(new RetryWhenNetWorkException()).subscribeOn(Schedulers.io()).unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread()).map(baseApi).subscribe(commonSubscriber);
        }

/**
 * 将原来的Observable转换为新的Observable，自定的新的Observable特性
 */
public void compose() {
    //  线程调度的经典引用

    //  创建一个线程控制的ObservableTransformer(可以将这个封装为工具类)
    Observable.Transformer<Object, Object> transformer = new Observable.Transformer<Object, Object>() {
        @Override
        public Observable<Object> call(Observable<Object> objectObservable) {
            return objectObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    };

    //  通过compose进行转换
    Observable.create(new Observable.OnSubscribe<Object>() {
        @Override
        public void call(Subscriber<? super Object> subscriber) {

        }
    }).compose(transformer);
}

private void initializeData() {

        application.getWebService()
                .getInteractionPosts(likeCounter, MySharedPreferences.getUserId(preferences), MySharedPreferences.getUserToken(preferences))
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .retryWhen(new RetryWithDelay(3,2000))
                .subscribe(new BaseSubscriber<PostSuperUserListModel>() {
                    @Override
                    public void onNext(PostSuperUserListModel response) {
                        showRecycleWithDataFilled(response);
                        progressFrame.setVisibility(View.GONE);
                    }
                    @Override
                    public void onError(Throwable e) {
                        progressFrame.setVisibility(View.GONE);
                        try {
                            Timber.e(e.getMessage());
                         //   Toast.makeText(UserLikeCounterActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
                        }catch (Exception ex){
                            ex.printStackTrace();
                        }
                    }
                });
    }

@Override
public void getVerifyCode() {
    Subscription subscription = Observable.just("")
            .subscribeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Action0() {
                @Override
                public void call() {
                    mLoginView.showLoading("获取验证码...");
                }
            })
            .observeOn(Schedulers.io())
            .subscribe(new Action1<Object>() {
                @Override
                public void call(Object o) {
                    mDataRepository.getVerifyCode("3", getIMEI(mLoginView.getViewContext()), mCallbackImp);
                }
            });

    mSubscriptions.add(subscription);
}

private void loadData() {
    Observable.just(DBManager.getInstance().getFriends())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(friends -> {
                if (friends != null && friends.size() > 0) {
                    mData.clear();
                    mData.addAll(friends);
                    getView().getFooterView().setText(UIUtils.getString(R.string.count_of_contacts, mData.size()));
                    //整理排序
                    SortUtils.sortContacts(mData);
                    if (mAdapter != null)
                        mAdapter.notifyDataSetChanged();
                }
            }, this::loadError);
}

public void getCategorys() {
    checkViewAttached();
    RxUtil.unsubscribe(mCategorySubscription);
    mCategorySubscription = mDataManager.getCategorys()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io()).subscribe(new Subscriber<CategoryResponse>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(CategoryResponse categoryResponse) {
                    getMvpView().showCategorys(categoryResponse);
                }
            });
}

private void getPassCode() {
    HttpService service = RetrofitManager.getInstance().getService();
    SubscriptionUtil.unSubscribe(subscription);
    subscription = service.getPassCode("login", "sjrand")
            .map(new Func1<ResponseBody, Bitmap>() {
                @Override
                public Bitmap call(ResponseBody responseBody) {
                    return BitmapFactory.decodeStream(responseBody.byteStream());
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Bitmap>() {
                @Override
                public void call(Bitmap bitmap) {
                    captcha.setImageBitmap(bitmap);
                }
            });
}

public static <T> Observable.Transformer<ApiResponse<T>, ApiResponse<T>> handleResult() {
    return new Observable.Transformer<ApiResponse<T>, ApiResponse<T>>() {
        @Override
        public Observable<ApiResponse<T>> call(Observable<ApiResponse<T>> tObservable) {
            return tObservable.flatMap(new Func1<ApiResponse<T>, Observable<ApiResponse<T>>>() {
                @Override
                public Observable<ApiResponse<T>> call(ApiResponse<T> response) {
                    if (response.state.code == 0) {
                        return createData(response);
                    } else {
                        return Observable.error(new ApiException(response.state));
                    }
                }
            }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    };
}

private void subscribeOn() {
    Observable<Integer> obs= Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            Log.d(TAG,"Observable:" + Thread.currentThread().getName());
            subscriber.onNext(1);
            subscriber.onCompleted();
        }
    });
    obs.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer integer) {
            Log.d(TAG,"Observer:" + Thread.currentThread().getName());
        }
    });
}

/**
 * 同步群组成员信息
 */
private void fetchGroupMembers() {
    mHasFetchedGroupMembers = false;
    Observable.from(getGroups())
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .subscribe(groups ->
                    ApiRetrofit.getInstance().getGroupMember(groups.getGroupId())
                            .subscribe(getGroupMemberResponse -> {
                                if (getGroupMemberResponse != null && getGroupMemberResponse.getCode() == 200) {
                                    List<GetGroupMemberResponse.ResultEntity> list = getGroupMemberResponse.getResult();
                                    if (list != null && list.size() > 0) {
                                        deleteGroupMembersByGroupId(groups.getGroupId());
                                        saveGroupMembers(list, groups.getGroupId());
                                    }
                                    mHasFetchedGroupMembers = true;
                                    checkFetchComplete();
                                } else {
                                    mHasFetchedGroupMembers = true;
                                    checkFetchComplete();
                                }
                            }, this::fetchGroupMembersError));
}

private Observable<Boolean> isStarred(String owner, final String repo) {
    return repoApi.checkIfRepoIsStarred(AccountHelper.getCurrentUid(), owner, repo)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .map(new Func1<Response, Boolean>() {
                @Override
                public Boolean call(Response response) {
                    /**
                     * Response if this repository is starred by you
                     *  Status: 204 No Content
                     * Response if this repository is not starred by you
                     *  Status: 404 Not Found
                     */
                    return response != null && response.code() == 204;
                }
            });
}

public void logout() {
    checkViewAttached();
    RxUtil.unsubscribe(mLogoutSubscription);
    mLogoutSubscription = mDataManager.logout()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<EmptyResponse>() {
                @Override
                public void onCompleted() {
                }

                @Override
                public void onError(Throwable e) {
                    Timber.e(e, "There was an error loading the DashBoardResponse.");
                    getMvpView().logoutError();
                }

                @Override
                public void onNext(EmptyResponse emptyResponse) {
                    getMvpView().logout();
                }
            });
}

@Override
protected void loadData() {
    int ver = 3;
    RetrofitHelper.getBiliAPI()
            .getVideoComment(mAid, mPageNum, mPageSize, ver)
            .compose(this.bindToLifecycle())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(videoCommentInfo -> {
                ArrayList<VideoCommentInfo.List> list = videoCommentInfo.list;
                ArrayList<VideoCommentInfo.HotList> hotList = videoCommentInfo.hotList;
                if (list.size() < mPageSize){
                    mLoadMoreView.setVisibility(View.GONE);
                    mAdapter.removeFootView();
                }
                mComments.addAll(list);
                mHotComments.addAll(hotList);
                finishTask();
            }, throwable -> {
                mLoadMoreView.setVisibility(View.GONE);
                mHeadView.setVisibility(View.GONE);
            });
}

private void getTags() {

    RetrofitHelper.getSearchAPI()
        .getHotSearchTags()
        .compose(bindToLifecycle())
        .map(HotSearchTag::getList)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(listBeans -> {

          hotSearchTags.addAll(listBeans);
          initTagLayout();
        }, throwable -> {

        });
  }

protected void sendUnlikeToServer(final VoicemeApplication application, int like, int hug, int same, int listen, final String message) {
    application.getWebService().unlikes(MySharedPreferences.getUserId(preferences), postId, like, hug, same, listen)
            .observeOn(AndroidSchedulers.mainThread())
            .retryWhen(new RetryWithDelay(3,2000))
            .subscribeOn(Schedulers.io())
            .subscribe(new BaseSubscriber<PostLikesResponse>() {
                @Override
                public void onNext(PostLikesResponse postLikesResponse) {
                }
                @Override
                public void onError(Throwable e) {
                    Crashlytics.logException(e);
                    try {
                        Toast.makeText(PostsDetailsActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
                    }catch (Exception ex){
                        ex.printStackTrace();
                    }
                }
            });
}

public void uploadPackageImage(Bitmap bitmap1, Bitmap bitmap2, Bitmap bitmap3) {

        mModel.uploadPackageImage(bitmap1, bitmap2, bitmap3)
                .subscribeOn(Schedulers.io())
                .retryWhen(new RetryWithDelay(3, 2))//遇到错误时重试,第一个参数为重试几次,第二个参数为重试的间隔
                .subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(AndroidSchedulers.mainThread())
                .compose(RxUtils.<BaseJson<JsonObject>>bindToLifecycle(mRootView))//使用RXlifecycle,使subscription和activity一起销毁
                .subscribe(new ErrorHandleSubscriber<BaseJson<JsonObject>>(mErrorHandler) {

                    @Override
                    public void onNext(BaseJson<JsonObject> jsonObjectBaseJson) {
                        if (jsonObjectBaseJson.getStatus() == 0) {
                            mRootView.packageImage(jsonObjectBaseJson.getStatus());
                        } else {
                            mRootView.showMessage(jsonObjectBaseJson.getMsgs());
                            mRootView.packageImage(jsonObjectBaseJson.getStatus());
                        }
                    }
                });
    }

public void getUser() {
    checkViewAttached();
    RxUtil.unsubscribe(mGetUserSubscription);
    mGetUserSubscription = mDataManager.getUserInfo().observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io()).subscribe(new Subscriber<UserInfoResponse>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(UserInfoResponse userInfoResponse) {
                    getMvpView().showUserInfo(userInfoResponse);
                }
            });
}

@Override
public void getBestComments(String bookReviewId) {
    Subscription rxSubscription = bookApi.getBestComments(bookReviewId)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<CommentList>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    LogUtils.e("getBestComments:" + e.toString());
                }

                @Override
                public void onNext(CommentList list) {
                    mView.showBestComments(list);
                }
            });
    addSubscrebe(rxSubscription);
}

/**
 * 同步群组信息
 */
private void fetchGroups() {
    mHasFetchedGroups = false;
    ApiRetrofit.getInstance().getGroups()
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .subscribe(getGroupResponse -> {
                if (getGroupResponse != null && getGroupResponse.getCode() == 200) {
                    List<GetGroupResponse.ResultEntity> list = getGroupResponse.getResult();
                    if (list != null && list.size() > 0) {
                        deleteGroups();
                        saveGroups(list);
                        //同步群组成员信息
                        fetchGroupMembers();
                    } else {
                        mHasFetchedGroupMembers = true;
                    }
                    mHasFetchedGroups = true;
                    checkFetchComplete();
                } else {
                    mHasFetchedGroups = true;
                    mHasFetchedGroupMembers = true;
                    checkFetchComplete();
                }
            }, this::fetchGroupsError);
}

@Override
protected void loadData() {

  RetrofitHelper.getRankAPI()
      .getAllareasRanks(type)
      .compose(bindToLifecycle())
      .map(allareasRankInfo -> allareasRankInfo.getRank().getList())
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(listBeans -> {

        allRanks.addAll(listBeans.subList(0, 20));
        finishTask();
      }, throwable -> {

        mSwipeRefreshLayout.setRefreshing(false);
        ToastUtil.shortToast("加载失败啦,请重新加载~");
      });
}

public void getBookDetail(String bookId) {
    Subscription rxSubscription = bookApi.getBookDetail(bookId).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<BookDetail>() {
                @Override
                public void onNext(BookDetail data) {
                    if (data != null && mView != null) {
                        mView.showBookDetail(data);
                    }
                }

                @Override
                public void onCompleted() {
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "onError: " + e);
                }
            });
    addSubscrebe(rxSubscription);
}

@Override
public void getHotReview(String book) {
    Subscription rxSubscription = bookApi.getHotReview(book).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<HotReview>() {
                @Override
                public void onNext(HotReview data) {
                    List<HotReview.Reviews> list = data.reviews;
                    if (list != null && !list.isEmpty() && mView != null) {
                        mView.showHotReview(list);
                    }
                }

                @Override
                public void onCompleted() {
                }

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

@Override
protected void loadData() {
    showRefreshing(true);
    final String code = getIntent().getStringExtra(KEY_EXTRA_CODE);
    subscription = RxLocation
            .get()
            .locate(this)
            .flatMap(new Func1<BDLocation, Observable<BaseBusResponse<BusLineStation>>>() {
                @Override
                public Observable<BaseBusResponse<BusLineStation>> call(BDLocation bdLocation) {
                    Map<String, String> options = new HashMap<>();
                    options.put("NoteGuid", code);
                    options.put("uid", BusController.uid);
                    options.put("DeviceID", BusController.deviceID);
                    options.put("sign", BusController.sign);
                    options.put("lat", String.valueOf(bdLocation.getLatitude()));
                    options.put("lng", String.valueOf(bdLocation.getLongitude()));
                    return ApiFactory.getBusController().getStationInfo(options).subscribeOn(Schedulers.io());
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<BaseBusResponse<BusLineStation>>() {
                @Override
                public void onCompleted() {
                    showRefreshing(false);
                }

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

                @Override
                public void onNext(BaseBusResponse<BusLineStation> response) {
                    adapter.setNewData(response.data.getList());
                }
            });
}

public void getGankDataFromInternet(final String postId) {
    NewsServer instance = NewsSingle.getInstance();
    Observable<Map<String, NewsDetailBean>> newDetail = instance.getNewDetail(postId);
    //没有相关推荐
    subscription = newDetail.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<Map<String, NewsDetailBean>>() {
                @Override
                public void onCompleted() {
                    Log.i(TAG, "onCompleted: ");
                }

                @Override
                public void onError(Throwable e) {
                    if (mLoadListener != null) {
                        if (e.toString().equals("java.lang.NullPointerException")) {
                            //没有相关推荐
                        } else {
                            mLoadListener.onNDError(new Exception(e.toString()));
                        }
                    }
                }

                @Override
                public void onNext(Map<String, NewsDetailBean> stringNewsDetailBeanMap) {
                    NewsDetailBean newsDetailBean = stringNewsDetailBeanMap.get(postId);
                    if (mLoadListener != null) {
                        mLoadListener.onNDSuccess(stringNewsDetailBeanMap.get(postId));
                    }
                }
            });

}

private void loadBookDetail() {
        DebugUtil.error("------http2");
        Subscription get = HttpClient.Builder.getDouBanService().getBookDetail(booksBean.getId())
//        Subscription get = HttpUtils.getInstance().getDouBanServer().getBookDetail(booksBean.getId())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<BookDetailBean>() {
                    @Override
                    public void onCompleted() {
                        showContentView();
                    }

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

                    @Override
                    public void onNext(final BookDetailBean bookDetailBean) {

                        mBookDetailUrl = bookDetailBean.getAlt();
                        mBookDetailName = bookDetailBean.getTitle();
                        bindingContentView.setBookDetailBean(bookDetailBean);
                        bindingContentView.executePendingBindings();
                    }
                });
        addSubscription(get);
    }

/**
 * 获取明星界面
 */
public void sellerDetails(HttpResultListener<Obj> listener, String id) {
    mApi.sellerDetails(id)
            .map(new HttpResultFuncNoList_has_obj())
            .subscribeOn(Schedulers.io())//在工作线程请求网络
            .observeOn(AndroidSchedulers.mainThread())//在主线程处理结果
            .subscribe(new HttpResultSubscriber<>(listener));
}