Espresso Wait for [RxIdlingResource] to Stop Standby

Question

I am new to testing espresso. In my existing application, we use RxAndroid to create a network. We use RxBus to communicate with parts of our application that would otherwise seem “impossible”.

We imported RxEspresso , which implements IdlingResourceso that we can use our RxAndroid network calls.

Unfortunately, RxEspresso does not allow RxBus to work, as it is "hot watched" and never closes. Therefore he throwsandroid.support.test.espresso.IdlingResourceTimeoutException: Wait for [RxIdlingResource] to become idle timed out

I made a small Android application demonstrating my point. It has two activities. The first shows some items received through a network call at startup RecyclerView.

When they click on it, it communicates via RxBus (I know that this is unnecessary, but just to demonstrate the point). Then DetailActivityshows the data.

How can we edit RxEspresso to work with our RxBus?

RxIdlingResource also validates RxEspresso

/**
 * Provides the hooks for both RxJava and Espresso so that Espresso knows when to wait
 * until RxJava subscriptions have completed.
 */

public final class RxIdlingResource extends RxJavaObservableExecutionHook implements IdlingResource {
    public static final String TAG = "RxIdlingResource";

    static LogLevel LOG_LEVEL = NONE;

    private final AtomicInteger subscriptions = new AtomicInteger(0);

    private static RxIdlingResource INSTANCE;

    private ResourceCallback resourceCallback;

    private RxIdlingResource() {
        //private
    }

    public static RxIdlingResource get() {
        if (INSTANCE == null) {
            INSTANCE = new RxIdlingResource();
            Espresso.registerIdlingResources(INSTANCE);
        }
        return INSTANCE;
    }

    /* ======================== */
    /* IdlingResource Overrides */
    /* ======================== */

    @Override
    public String getName() {
        return TAG;
    }

    @Override
    public boolean isIdleNow() {
        int activeSubscriptionCount = subscriptions.get();
        boolean isIdle = activeSubscriptionCount == 0;

        if (LOG_LEVEL.atOrAbove(DEBUG)) {
            Log.d(TAG, "activeSubscriptionCount: " + activeSubscriptionCount);
            Log.d(TAG, "isIdleNow: " + isIdle);
        }

        return isIdle;
    }

    @Override
    public void registerIdleTransitionCallback(ResourceCallback resourceCallback) {
        if (LOG_LEVEL.atOrAbove(DEBUG)) {
            Log.d(TAG, "registerIdleTransitionCallback");
        }
        this.resourceCallback = resourceCallback;
    }

    /* ======================================= */
    /* RxJavaObservableExecutionHook Overrides */
    /* ======================================= */

    @Override
    public <T> Observable.OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance,
                                                          final Observable.OnSubscribe<T> onSubscribe) {
        int activeSubscriptionCount = subscriptions.incrementAndGet();
        if (LOG_LEVEL.atOrAbove(DEBUG)) {
            if (LOG_LEVEL.atOrAbove(VERBOSE)) {
                Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount, new Throwable());
            } else {
                Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount);
            }
        }

        onSubscribe.call(new Subscriber<T>() {
            @Override
            public void onCompleted() {
                onFinally(onSubscribe, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                onFinally(onSubscribe, "onError");
            }

            @Override
            public void onNext(T t) {
                //nothing
            }
        });

        return onSubscribe;
    }



    private <T> void onFinally(Observable.OnSubscribe<T> onSubscribe, final String finalizeCaller) {
        int activeSubscriptionCount = subscriptions.decrementAndGet();
        if (LOG_LEVEL.atOrAbove(DEBUG)) {
            Log.d(TAG, onSubscribe + " - " + finalizeCaller + ": " + activeSubscriptionCount);
        }
        if (activeSubscriptionCount == 0) {
            Log.d(TAG, "onTransitionToIdle");
            resourceCallback.onTransitionToIdle();
        }
    }
}

Rxbus

public class RxBus {

    //private final PublishSubject<Object> _bus = PublishSubject.create();

    // If multiple threads are going to emit events to this
    // then it must be made thread-safe like this instead
    private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());

    public void send(Object o) {
        _bus.onNext(o);
    }

    public Observable<Object> toObserverable() {
        return _bus;
    }

    public boolean hasObservers() {
        return _bus.hasObservers();
    }
}

Mainactivity

public class MainActivity extends AppCompatActivity {

    @Bind(R.id.rv)
    RecyclerView RV;

    private List<NewsItem> newsItems;
    private RecyclerViewAdapter adapter;
    private Observable<List<NewsItem>> newsItemsObservable;
    private CompositeSubscription subscriptions = new CompositeSubscription();
    private RxBus rxBus;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ButterKnife.bind(this);

        //Subscribe to RxBus
        rxBus = new RxBus();
        subscriptions.add(rxBus.toObserverable()
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object event) {
                        //2.
                        NewsItem myClickNewsItem = (NewsItem) event;
                        startActivity(new Intent(MainActivity.this, DetailActivity.class).putExtra("text", myClickNewsItem.getBodyText()));
                    }
                }));

        //Set the adapter
        adapter = new RecyclerViewAdapter(this);
        //Set onClickListener on the list
        ItemClickSupport.addTo(RV).setOnItemClickListener(new ItemClickSupport.OnItemClickListener() {
            @Override
            public void onItemClicked(RecyclerView recyclerView, int position, View v) {
                //Send the clicked item over the RxBus.
                //Receives it in 2.
                rxBus.send(newsItems.get(position));
            }
        });
        RV.setLayoutManager(new LinearLayoutManager(this));
        RestAdapter retrofit = new RestAdapter.Builder()
                .setEndpoint("http://URL.com/json")
                .build();
        ServiceAPI api = retrofit.create(ServiceAPI.class);
        newsItemsObservable = api.listNewsItems(); //onComplete goes to setNewsItems
    }

    @Override
    protected void onPostCreate(Bundle savedInstanceState) {
        super.onPostCreate(savedInstanceState);
        NewsItemObserver observer = new NewsItemObserver(this);
        newsItemsObservable.delaySubscription(1, TimeUnit.SECONDS).observeOn(AndroidSchedulers.mainThread()).subscribeOn(Schedulers.io()).subscribe(observer);
    }

    public void setNewsItems(List<NewsItem> newsItems) {
        this.newsItems = newsItems;
        adapter.setNewsItems(newsItems);
        RV.setAdapter(adapter);
    }

Answer 1

, , objects RxBus subscriptions.incrementAndGet();

. SerializedSubject PublishSubject.

, .

@Override
public <T> Observable.OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final Observable.OnSubscribe<T> onSubscribe) {

    int activeSubscriptionCount = 0;

    if (observableInstance instanceof SerializedSubject || observableInstance instanceof PublishSubject) {
        Log.d(TAG, "Observable we won't register: " + observableInstance.toString());
    } else {
        activeSubscriptionCount = subscriptions.incrementAndGet();
    }
    if (LOG_LEVEL.atOrAbove(DEBUG)) {
        if (LOG_LEVEL.atOrAbove(VERBOSE)) {
            Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount, new Throwable());
        } else {
            Log.d(TAG, onSubscribe + " - onSubscribeStart: " + activeSubscriptionCount);
        }
    }


    onSubscribe.call(new Subscriber<T>() {
        @Override
        public void onCompleted() {
            onFinally(onSubscribe, "onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            onFinally(onSubscribe, "onError");
        }

        @Override
        public void onNext(T t) {
            Log.d(TAG, "onNext:: " + t.toString());
            //nothing
        }
    });

    return onSubscribe;
}