AsyncTask in RxJava

Question

I cannot figure out how to translate a simple AsyncTask to 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();
    }
}

How can this be translated in the same operational state of RxJava? Thanks!

Answer 1

Instead of using it, Observable.createyou should use either Observable.defer()or else Observable.fromCallable(which was introduced in RxJava 1.0.15) - because these methods will provide the correct observable contract and save you some errors that you can enter when creating observables manually.

, runOnUiThread, , AndroidSchedulers.mainThread(), . RxAndroid , .

:

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 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)));

Answer 2

, , ,

• onNext
• onError
• OnCompleted

"".

rx , , , . , , , , .

- , :

    public Observable<String> getProxyAsync() {
        return Observable.create(subscriber -> {
            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();

                subscriber.onNext(proxy_arr);
            } catch (IOException e) {
                subscriber.onError(e);
            } finally {
                subscriber.onCompleted();
            }
        });
    }

, - , :

public void myPreciousMethod() {
    myCustomRepo.getProxyAsync()
        .subscribeOn(Schedulers.newThread())
        .subscribe(result -> {
            runOnUiThread(() -> {
                if (result.equals("err_internet")) {
                    func.toastMessage(R.string.toast_err_nointernet, "", "alert");
                }
            });
        }, throwable -> {
            // some exception happened emmited by your code, handle it well
            new DebugLog(getActivity(), "News", "Sync PROXY", Log.getStackTraceString(e));
        }, () -> {
            // onCompleted:
            runOnUiThread(() -> reloadAdapter());
        });
}

.runOnUiThread() ( rx), , . ( .observeOn() .subscribeOn()). retrolambda .

Answer 3

This is one way to do this. You can opt out of deferral if necessary.

        Observable.defer(new Func0<Observable<String>>() {
                @Override
                public Observable<String> call() {
                    return Observable.create(new Observable.OnSubscribe<String>() {
                        @Override
                        public void call(Subscriber<? super String> subscriber) {
                            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));
                            }
                            if (!subscriber.isUnsubscribed()) {
                                subscriber.onNext(proxy_arr);
                            }
                        }
                    })
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<String>() {
                @Override
                public void call(String result) {
                    if (result.equals("err_internet")){
                        func.toastMessage(R.string.toast_err_nointernet, "", "alert");
                    }
                    reloadAdapter();
                }
            });