Verify that method is called in onNext of RxJava Subscriber

Question

I have the following method that uses a Retrofit service interface to fetch some data from an API, then interacts with a view interface.

@Override
@VisibleForTesting
public void fetchPhotos(@Nullable PhotosService service, @Nullable Scheduler subscribeOn) {
    view.showLoading();

    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(Constants.PLACEHOLDER_API_BASE_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

    if (service == null) service = retrofit.create(PhotosService.class);
    if (subscribeOn == null) subscribeOn = Schedulers.newThread();

    service.listPhotos()
            .subscribeOn(subscribeOn)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(photoList -> {
                Log.d(TAG, "got photos " + photoList.toString());
                view.unshowLoading();
            }, throwable -> {
                Log.d(TAG, "error " + throwable.toString());
                view.unshowLoading();
                view.displayError(throwable.toString(), v -> fetchPhotos());
            });
}

I want to test that view.unshowLoading() is called in onNext.

Here's my test:

@Test
public void viewUnshowsLoadingAfterFetchingPhotos() {

    PhotosListView view = Mockito.mock(PhotosListView.class);

    PhotosListPresenter presenter = new PhotosListPresenterImpl(view);

    presenter.fetchPhotos(() -> Observable.create(new Observable.OnSubscribe<List<Photo>>() {
        @Override
        public void call(Subscriber<? super List<Photo>> subscriber) {

            subscriber.onNext(new ArrayList<Photo>());

        }
    }), Schedulers.immediate());

    Mockito.verify(view).unshowLoading();


}

I explicitly pass in the Scheduler Schedulers.immediate() to make sure onNext() is called instantly on the subscribing thread.

When I debug through my method though, onNext() is not called. What am I doing wrong or how could I best test this?

EDIT: This article brought me on to something:

If you wish to change the thread on which the operation is performed you can call subscribeOn(). To get back to the main thread use observeOn(AndroidSchedulers.mainThread()). However, notice that whenever you force the operation onto a specific thread, it will always make the subscription asynchronous.

When I omit the

        .subscribeOn(subscribeOn)
        .observeOn(AndroidSchedulers.mainThread())

part, the test works as expected. I've rearranged my method to no call in observeOn() or subscribeOn() when no schedulers are passed in:

public void fetchPhotos(@Nullable PhotosService service, @Nullable Scheduler subscribeOn, @Nullable Scheduler observeOn) {
    view.showLoading();

    if (service == null) service = createService();

    Observable<List<Photo>> observable = service.listPhotos();

    if (subscribeOn != null) observable = observable.subscribeOn(subscribeOn);
    if (observeOn != null) observable = observable.observeOn(observeOn);

    observable.subscribe(photoList -> {
        Log.d(TAG, "got photos " + photoList.toString());
        view.unshowLoading();
    }, throwable -> {
        Log.d(TAG, "error " + throwable.toString());
        view.unshowLoading();
        view.displayError(throwable.toString(), v -> fetchPhotos());
    });
}

Looks a bit clumsy, but works.

Any ideas still welcome :)

Have you tried a different Scheduler other than `immediate`? — IgorGanapolsky, Nov 18 '15 at 17:57
@IgorGanapolsky I've also tried AndroidSchedulers.mainthread() — fweigl, Nov 18 '15 at 17:58
In your presenter you have `Schedulers.newThread()`, and in your test you have `Schedulers.immediate())`. I wonder if they are conflicting somehow... — IgorGanapolsky, Nov 18 '15 at 18:05
@IgorGanapolsky I've verified by debugging that the ImmediateScheduler is used during the test. — fweigl, Nov 18 '15 at 18:06
But `newThread` is for scheduling long running or blocking actions. Can you try replacing it to see if it solves your issue? — IgorGanapolsky, Nov 18 '15 at 18:22
@IgorGanapolsky newThread is not used when the ImmediateScheduler is passed in. I removed the newThread part / switched it to Schedulers.immediate(), doesn't make a difference. — fweigl, Nov 18 '15 at 18:34
What is the output? Perhaps the main thread terminates since you're using `subscribeOn`? — Reut Sharabani, Nov 19 '15 at 07:10
@Reut There is no output on the test described in the first part of my question. The test terminates before onNext() is called. And yes, I think subscribeOn() and observeOn() forced the subscription to another thread, as outlined in the citation i posted. — fweigl, Nov 19 '15 at 07:16
Why don't you try using a countdownlatch to make sure the main thread waits for the observable? Add a `CountDownLatch` and use `doOnTerminate(() -> countDownLatch.countDown())` and see what happens. — Reut Sharabani, Nov 19 '15 at 07:17

score 1 · Answer 1 · answered Nov 20 '15 at 16:31

The first sample is fine, just inject the ui scheduler and use that. In your test inject something like an immediate scheduler, and in production inject the Android ui scheduler. In general it's better not to hardcode dependencies in your classes, but rather to inject them. This is one of those cases in which dependency injection could have helped.

A note on subscribeOn: you don't need to use it with retrofit as retrofit will perform the operation on a default thread anyway. Also naming schedulers "subscribeOn" and "observeOn" doesn't make much sense, as you might want to use the same scheduler to pass to subscribeOn() and observeOn(). It would be better to give them more meaningful names given what they represent, for example "backgroundScheduler" and "uiScheduler"

Verify that method is called in onNext of RxJava Subscriber

1 Answers1