RxJava subscriber as a parameter does not work, new identical subscriber work

Question

We're using rxandroid:1.1.0 along with rxjava:1.1.2 in our application. Our project has the RxJava Subscriber created this way:

• Android Activity creates the Subscriber. When user deletes an item we pass the Subscriber as argument to a "network service" class we have
• The network service class sets up the Observable, and gets the Subscriber as argument. The Observable is set to subscribe to the Subscriber argument

This unfortunately doesn't work. The subscriber never gets called and we don't get errors or any more information.

However, if instead of using the argument Subscriber, and I create a brand new Subscriber of the same time, the Observable works just fine.

Any idea how I could make this work with the argument Subscriber? We want to call UI methods onComplete from the Activity that are not accessible from the network service class.

Here's my activity code:

public Subscriber<Models.FavoriteResponse> mFavoritesSubscriber = new Subscriber<Models.FavoriteResponse>() {
    @Override
    public void onCompleted() {
        // empty
        Logger.w("mFavoritesSubscriber onCompleted");
        showFavorites();
    }

    @Override
    public void onError(Throwable e) {
        Logger.w("mFavoritesSubscriber onError");
    }

    @Override
    public void onNext(Models.FavoriteResponse favoriteResponse) {
        Logger.w("mFavoritesSubscriber onNext");
        getHelper().setIsFavoriteOnProducts(favoriteResponse.getFavoriteResponseList());
    }
};

private void removeSelectedFavorites() {
    for (Product selectedProduct : selectedProducts) {
        ServiceUtils.removeFavorite(this, selectedProduct.sku, mFavoritesSubscriber);
    }
}

And here's the code from my network service class. This version does not work:

public static void removeFavorite(final Context context, final String sku, Subscriber<Models.FavoriteResponse> subscriber) {

    Logger.w("ServiceUtils removeFavorite");

    String userId = PersistentSharedPreferences.readUserId(context);
    FavoritesServices service = FavoritesServices.Companion.create(BuildConfig.ORDER_SERVICE_DOMAIN,
            new Interceptor(context));
    final Observable<Models.FavoriteResponse> observable = service
            .deleteFavorite(userId, sku, getFavoritesParameters(context));

    observable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .onErrorResumeNext(new Func1<Throwable, Observable<? extends Models.FavoriteResponse>>() {
                @Override
                public Observable<? extends Models.FavoriteResponse> call(Throwable throwable) {
                    Logger.w("ServiceUtils onErrorResumeNext");
                    return null;
                }
            })
            .subscribe(subscriber);
}

But this version of the network service class method works fine. Notice the new Subscriber:

public static void removeFavorite(final Context context, final String sku, Subscriber<Models.FavoriteResponse> subscriber) {

    Logger.w("ServiceUtils removeFavorite");

    String userId = PersistentSharedPreferences.readUserId(context);
    FavoritesServices service = FavoritesServices.Companion.create(BuildConfig.ORDER_SERVICE_DOMAIN,
            new Interceptor(context));
    final Observable<Models.FavoriteResponse> observable = service
            .deleteFavorite(userId, sku, getFavoritesParameters(context));

    observable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .onErrorResumeNext(new Func1<Throwable, Observable<? extends Models.FavoriteResponse>>() {
                @Override
                public Observable<? extends Models.FavoriteResponse> call(Throwable throwable) {
                    Logger.w("ServiceUtils onErrorResumeNext");
                    return null;
                }
            })
            .subscribe(new Subscriber<Models.FavoriteResponse>() {
                @Override
                public void onCompleted() {
                    Logger.w("ServiceUtils Subscriber onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    Logger.w("ServiceUtils Subscriber onError");
                }

                @Override
                public void onNext(Models.FavoriteResponse favoriteResponse) {
                    Logger.w("ServiceUtils Subscriber onNext");
                }
            });
}

Any help would be appreciated. Thank you!

Answer 1

Don't reuse a Subscriber. Subscriber is a Subscription and it's stateful. Subscriber won't be notified after it's unsubscribed.

You can use Observer instead.