RxJava PublishProcessor does not receive onNext

Question

I'am fairly new to RxJava and try to build up the Model View Intent Pattern in Android. In my View (Activity) i create a PublishProcessor as follows:

private PublishProcessor<MviResultIntent> mPublishProcessor;
mPublishProcessor = PublishProcessor.create();

After the creation I'am calling a method of my presenter with the Processor as a Parameter:

mResultPresenter.bindIntents(mPublishProcessor);

What happens inside the called method:

        Disposable processIntents = mPublishProcessor 
            .subscribeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableSubscriber<MviResultIntent>() {

                @Override
                public void onNext(MviResultIntent mviResultIntent) {
                }

                @Override
                public void onError(Throwable e) {
                }

                @Override
                public void onComplete() {
                    //may be ignored
                }
            });

    mCompositeDisposable.add(processIntents);

and in my View Class i call afterwards:

mPublishProcessor.onNext(new MviResultIntent.ProductsIntent());

The PublishProcessor inside my Presenter does not get the onNext Event I'am trying to trigger. Am i missing something? I dont receive onComplete or onError neither.

Any help is appreciated! If you need any further Information feel free to ask. Thanks in advance for your help!

Check mPublishProcessor.hasSubscribers() before onNext to verify there are actually subscriber(s) listening the time. — akarnokd
As you expected there are no Subscribers when i call the onNext but i have no solution by far. — Lars
You have to call that subscribeWith way before the event is fired. — akarnokd
Yes thats what im doing or at least i believe i do. I call the creation of the PublishProcessor and the bindIntents Method inside onCreate and the onNext inside onResume. — Lars
I have another similar construct where it actually works. But the hasSubscribers returns false there too. Im on the Version 2.1.10 if this might be a hint — Lars

Lars Lars · Accepted Answer · 2018-04-13T14:13:33

I figured out that i can't subscribe inside a non Android class on the Main Thread. When i change the

.subscribeOn(AndroidSchedulers.mainThread())

to

.subscribeOn(Schedulers.io())

it works.

RxJava PublishProcessor does not receive onNext

1 Answers