Changing AsyncTask to Rxjava

Issue

I have a challenge with converting the AsyncTask doInBackground process to RxJava. I would love to know how to convert this to Rx Java as none of what I've tried is working.

new AsyncTask<Void, Void, Integer>() {
            @Override
            protected Integer doInBackground(Void... voids) {

                return mDAO.getCount();
            }

            @Override
            protected void onPostExecute(Integer count) {
                if (count == 0)
                    mCount.setText("All Notifications");
                else
                    mCount.setText("New Notificaiton "+count);
            }
        }.execute();

And I tried this for Rx

Observable<Integer> count = Observable.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return mDAO.getCount();
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
        count.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                mDisposable.add(d);
            }

            @Override
            public void onNext(Integer integer) {
                if (integer == 0)
                    mCount.setText("All Notifications");
                else
                    mCount.setText("New Notification "+count);

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

I get this for instead of the count

Count io.reactivex.internal.operators.observable.ObservableObserveOn@5ccee5b

How do I solve this? Thank you.

Solution

Your implementation is why you're having this error. You should use a single callable instead. This should work 100% and let me know if you have any challenge with it.

Single.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return mDAO.getCount();
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSuccess(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        if (integer == 0)
                            mCount.setText("All Notifications");
                        else
                            mCount.setText("New Notification "+integer);
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {

                    }
                })
                .subscribe();

Answered By - Mbuodile Obiosio