AsyncTask in RxJava

Question

I can not understand how to translate a simple AsyncTask in RxJava. Take for example:

private class Sync extends AsyncTask<String, String, String> {

    @Override
    protected String doInBackground(String... params) {
              String proxy_arr = "";
                    try {
                        Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST)
                                .userAgent(Constants.USER_AGENT)
                                .ignoreContentType(true)
                                .ignoreHttpErrors(true)
                                .timeout(Constants.USER_TIMEOUT)
                                .get();

                        if (jsoup_proxy != null) proxy_arr = jsoup_proxy.text().trim();
                    } catch (IOException e) {
                        new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(e));
                    }
              return proxy_arr;
    }

    @Override
    protected void onPostExecute(String result) {
        if (result.equals("err_internet")){
            func.toastMessage(R.string.toast_err_nointernet, "", "alert");
        }

        reloadAdapter();
    }
}

As it can be translated in the same working condition RxJava? Thank you!

Answer 1

Instead of using Observable.create you should use either Observable.defer() or better yet Observable.fromCallable (which was introduced in RxJava 1.0.15) - because these methods will ensure a proper observable contract and save you from some mistakes you can introduce when creating observable by hand.

Also instead of going with runOnUiThread as suggested in one of the answers above, you should really use AndroidSchedulers.mainThread() which was created for exactly this purpose. Just use RxAndroid library which provides it.

I suggest the following solution:

public Observable<String> getJsoupProxy() {
  return Observable.fromCallable(() -> {
      try {
        Document jsoup_proxy = Jsoup.connect(Constants.SITE_PROXY_LIST)
          .userAgent(Constants.USER_AGENT)
          .ignoreContentType(true)
          .ignoreHttpErrors(true)
          .timeout(Constants.USER_TIMEOUT)
          .get();

        return jsoup_proxy != null ? jsoup_proxy.text().trim() : "";
      } catch (IOException e) {
        // just rethrow as RuntimeException to be caught in subscriber's onError
        throw new RuntimeException(e);
      }
    });
}

getJsoupProxy()
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread()) // this scheduler is exported by RxAndroid library
  .subscribe(
     proxy -> {
       if(proxy.equals("err_internet")) {
         // toast
       }
       reloadAdapter();
     },
     error -> new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(error)));