Proposal to add RxIdlingResource for Espresso testing

[rosshambrick]

Problem: Espresso does not know about RxJava's async mechanisms, so it doesn't wait for subscriptions to finish before continuing on to the next testing step. Thereby, Espresso testing when using RxJava does not work out-of-the-box.

Solution: I've created a RxIdlingResource class that tracks subscriptions via a RxJavaObservableExecutionHook plugin and is registered with Espresso as an IdlingResource. This gives Espresso the knowledge it needs to wait or proceed with respect to RxJava background processes.

Also, since registering the plugin has to be done at a specific time in the startup process, I've created an RxAndroidJUnitRunner class that takes care of the setup.

Proposal Since this project seems to fill the gaps between RxJava and the standard Android api's, it seems reasonable to add RxIdlingResource to fill the gap between RxJava and the Espresso testing framework. If everyone agrees that this belongs in the RxAndroid project I can go ahead and create a PR.

[kboyarshinov]

+1. I have also implemented this. Would be nice to have RxAndroid solution.

[JakeWharton]

This would have to be a sibling module for the dependency on Espresso. Also since the Nebula plugins prevent us from actually making Android library modules we probably can't even do this.

[rosshambrick]

Agreed on the separate module. I'm not familiar with Nebula so I can't speak to it's feasibility. I'll do a little research for my own benefit, but if it is in fact not going to be a possibility, I'll just create an new github project for this.

@kboyarshinov I've found an implementation that works for my own testing scenario but I'm not yet confident that it is a sound general purpose solution. So, no matter where this lands, I'd like your input so that the implementation can be as widely useful as possible.

[kboyarshinov]

@rosshambrick Great. I'll be glad to contribute.

[rosshambrick]

@kboyarshinov (and others) I've created an new repo RxEspresso to provide this functionality. Please feel free to provide feedback and/or contribute.

I'm closing this issue now.

[cxzhang2]

Seems this issue has still not been addressed. The solution initially proposed here and implemented in https://github.com/stablekernel/RxEspresso equates Espresso's idle state to the end of a subscription, which is fairly limiting as RxEspresso's documentation points out.

@tommyd3mdi in stablekernel/RxEspresso#4 has suggested an alternative solution: wrapping the action that is passed to onSchedule(). However, this is also not a complete solution as it does not support actions that are recursively scheduled through schedulePeriodically(). In fact a stracktrace lookup to identify schedulePeriodically() by name is required just to avoid this case.

Are we doomed to have Espresso code live in /main forever? Is there no way for to hook into Rx from /androidTest?

[artem-zinnatullin]

There is a way, but it requires custom scheduler. Basically you need to increment counter when action is scheduled and decrement when it's done. You can put this schedulers to tests and replace real ones with hook, so no main sources need to be modified.

On Thu, 21 Jul 2016, 06:32 Chris Zhang, notifications@github.com wrote:

Seems this issue has still not been addressed. The solution initially proposed here and implemented in https://github.com/stablekernel/RxEspresso equates Espresso's idle state to the end of a subscription, which is fairly limiting as RxEspresso's documentation points out.

@tommyd3mdi https://github.com/tommyd3mdi in stablekernel/RxEspresso#4 https://github.com/stablekernel/RxEspresso/issues/4 has suggested an alternative solution: wrapping the action that is passed to onSchedule(). However, this is also not a complete solution as it does not support actions that are recursively scheduled through schedulePeriodically(). In fact a stracktrace lookup to identify schedulePeriodically() by name is required just to avoid this case.

Are we doomed to have Espresso code live in /main forever? Is there no way for to hook into Rx from /androidTest?

— You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub https://github.com/ReactiveX/RxAndroid/issues/149#issuecomment-234147778, or mute the thread https://github.com/notifications/unsubscribe-auth/AA7B3EymPANMgd2VB7ZcQizzQkbLPhYhks5qXuhFgaJpZM4DymfO .

[cxzhang2]

Of course! Thank you again artem!

[cxzhang2]

Whipped up a little class as per @artem-zinnatullin's suggestions, would love some feedback:

public class IdlingScheduler extends Scheduler {

    private CountingIdlingResource countingIdlingResource;

    private Scheduler scheduler;

    public IdlingScheduler(Scheduler scheduler) {
        this.scheduler = scheduler;
        String resourceName = scheduler.getClass().getSimpleName() + scheduler.hashCode();
        countingIdlingResource = new CountingIdlingResource(resourceName, true);
    }

    @Override
    public Worker createWorker() {
        return new IdlingWorker(scheduler.createWorker());
    }

    public CountingIdlingResource countingIdlingResource() {
        return countingIdlingResource;
    }

    private class IdlingWorker extends Worker {

        private Worker worker;
        private boolean recursive;

        public IdlingWorker(Worker worker) {
            this.worker = worker;
        }

        @Override
        public Subscription schedule(Action0 action) {
            return recursive ?
                    worker.schedule(action) :
                    worker.schedule(decorateAction(action));
        }

        @Override
        public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
            return recursive ?
                    worker.schedule(action, delayTime, unit) :
                    worker.schedule(decorateAction(action), delayTime, unit);
        }

        @Override
        public Subscription schedulePeriodically(Action0 action, long initialDelay, long period, TimeUnit unit) {
            recursive = true;
            return worker.schedulePeriodically(decorateAction(action), initialDelay, period, unit);
        }

        @Override
        public void unsubscribe() {
            worker.unsubscribe();
        }

        @Override
        public boolean isUnsubscribed() {
            return worker.isUnsubscribed();
        }

        private Action0 decorateAction(Action0 action) {
            return () -> {
                countingIdlingResource.increment();
                try {
                    action.call();
                } finally {
                    countingIdlingResource.decrement();
                }
            };
        }
    }
}

edit: using CountingIdlingResource per @JakeWharton's recommendation, much nicer. edit2: now using unique resource names for CountingIdlingResource, thanks @Maragues

[JakeWharton]

Consider CountingIdlingResource which does most of that for you.

[Maragues]

@cxzhang2 And how would that register/unregister in Espresso? Wouldn't you need 1 CountingIdlingResource per Scheduler? If that's the case, you'd better instantiate them with different names instead of the TAG constant

Also if you could provide an example on how to hook your schedulers. What I use for JUnit tests doesn't work out of the box.

[artem-zinnatullin]

You need to register this hook before any RxJava invocations otherwise it may lead to inconsistent counter state. To register hook I'd suggest to use custom instrumentation test runner.

On Fri, 22 Jul 2016, 15:25 Miguel Aragues, notifications@github.com wrote:

@cxzhang2 https://github.com/cxzhang2 And how would that register/unregister in Espresso? Wouldn't you need 1 CountingIdlingResource per Scheduler? If that's the case, you'd better instantiate them with different names instead of the TAG constant

Also if you could provide an example on how to hook your schedulers. What I use for JUnit tests doesn't work out of the box.

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/ReactiveX/RxAndroid/issues/149#issuecomment-234530107, or mute the thread https://github.com/notifications/unsubscribe-auth/AA7B3PzcG5p6eVOe8skUmkJ37SIqHQpuks5qYLakgaJpZM4DymfO .

[cxzhang2]

@Maragues good call! Fixed.

Usage:

1. In /androidTest, instantiate IdlingScheduler idlingScheduler = new IdlingScheduler(your production scheduler here)
2. inject idlingScheduler wherever you were injecting your production scheduler
3. register idlingScheduler.countingIdlingResource() with Espresso.

See also Artem's suggestion about the custom instrumentation test runner. Right now I'm launching a test version of my Application class using a custom runner. The testApp overrides my component preparation method and feeds IdlingScheduler into Dagger in place of my production scheduler.

[Maragues]

@artem-zinnatullin Do you have an example of a custom runner that does that? I've been trying for the last hour without success :(

public class CustomTestRunner extends AndroidJUnitRunner {

  private final RxJavaSchedulersHook mRxJavaSchedulersHook = new RxJavaSchedulersHook() {
    private IdlingScheduler ioIdlingScheduler, computationIdlingScheduler;

    @Override
    public Scheduler getComputationScheduler() {
      if (computationIdlingScheduler == null)
        computationIdlingScheduler = convertToIdlingScheduler(super.getComputationScheduler());

      Log.d(TAG,"Custom getComputationScheduler");
      return computationIdlingScheduler;
    }

    @Override
    public Scheduler getIOScheduler() {
      if (ioIdlingScheduler == null)
        ioIdlingScheduler = convertToIdlingScheduler(super.getIOScheduler());

      return ioIdlingScheduler;
    }
  };

  private final RxAndroidSchedulersHook mRxAndroidSchedulersHook = new RxAndroidSchedulersHook() {
    private IdlingScheduler mainthreadIdlingScheduler;

    @Override
    public Scheduler getMainThreadScheduler() {
      if (mainthreadIdlingScheduler == null)
        mainthreadIdlingScheduler = convertToIdlingScheduler(super.getMainThreadScheduler());

      Log.d(TAG,"Custom getMainThreadScheduler");
      return mainthreadIdlingScheduler;
    }
  };

  public void onCreate(Bundle arguments) {
    super.onCreate(arguments);

    Log.d(TAG,"Custom oncreate");
    hookSchedulers();
  }

  void hookSchedulers() {
    try {
      RxAndroidPlugins.getInstance().reset();
      RxAndroidPlugins.getInstance().registerSchedulersHook(mRxAndroidSchedulersHook);
      //getInstance is deprecated, but it works in JUnit tests
      RxJavaPlugins.getInstance().registerSchedulersHook(mRxJavaSchedulersHook);

      RxAndroidPlugins.getInstance().reset();
    } catch (InvocationTargetException e) {
      e.printStackTrace();
    } catch (IllegalAccessException e) {
      e.printStackTrace();
    } catch (NoSuchMethodException e) {
      e.printStackTrace();
    }
  }
}

Logcat prints "Custom onCreate", but "Custom getMainThreadScheduler" is never printed. Plus, I can't debug the runner, I guess it's created before debugging is possible.

Thanks!

[JakeWharton]

final class EspressoRxJavaSchedulersHook extends RxJavaSchedulersHook {
  @Override public Scheduler getComputationScheduler() {
    Scheduler delegate = createComputationScheduler();
    DelegatingIdlingResourceScheduler wrapped =
        new DelegatingIdlingResourceScheduler(delegate, "RxJava Computation Scheduler");
    Espresso.registerIdlingResources(wrapped);
    return wrapped;
  }

  @Override public Scheduler getIOScheduler() {
    Scheduler delegate = createIoScheduler();
    DelegatingIdlingResourceScheduler wrapped =
        new DelegatingIdlingResourceScheduler(delegate, "RxJava I/O Scheduler");
    Espresso.registerIdlingResources(wrapped);
    return wrapped;
  }

  @Override public Scheduler getNewThreadScheduler() {
    Scheduler delegate = createNewThreadScheduler();
    DelegatingIdlingResourceScheduler wrapped =
        new DelegatingIdlingResourceScheduler(delegate, "RxJava New Thread Scheduler");
    Espresso.registerIdlingResources(wrapped);
    return wrapped;
  }
}

[artem-zinnatullin]

What Jake posted is similar to what I had.

Though for now in the project I work on we decided to use condition watcher and "wait" for actual UI conditions (through idling API) instead of letting tests know how async operations work under the hood.

On Fri, 22 Jul 2016, 18:43 Jake Wharton, notifications@github.com wrote:

final class EspressoRxJavaSchedulersHook extends RxJavaSchedulersHook { @Override public Scheduler getComputationScheduler() { Scheduler delegate = createComputationScheduler(); DelegatingIdlingResourceScheduler wrapped = new DelegatingIdlingResourceScheduler(delegate, "RxJava Computation Scheduler"); Espresso.registerIdlingResources(wrapped); return wrapped; }

@Override public Scheduler getIOScheduler() { Scheduler delegate = createIoScheduler(); DelegatingIdlingResourceScheduler wrapped = new DelegatingIdlingResourceScheduler(delegate, "RxJava I/O Scheduler"); Espresso.registerIdlingResources(wrapped); return wrapped; }

@Override public Scheduler getNewThreadScheduler() { Scheduler delegate = createNewThreadScheduler(); DelegatingIdlingResourceScheduler wrapped = new DelegatingIdlingResourceScheduler(delegate, "RxJava New Thread Scheduler"); Espresso.registerIdlingResources(wrapped); return wrapped; } }

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/ReactiveX/RxAndroid/issues/149#issuecomment-234579344, or mute the thread https://github.com/notifications/unsubscribe-auth/AA7B3LA0PWhtmSs50qmTJJnBPpUHcQrJks5qYOU-gaJpZM4DymfO .

[msgitter]

Thx all for sharing your ideas!

@cxzhang2 I think there is subtle bug in your decorateAction method. The increment should be called immediately, and not in the returned action.

        private Action0 decorateAction(Action0 action) {
            countingIdlingResource.increment();
            return () -> {
                try {
                    action.call();
                } finally {
                    countingIdlingResource.decrement();
                }
            };
        }

@Maragues Why are you calling reset again after calling registerSchedulerHook?

[Maragues]

@marksutt That code wasn't working, maybe that's one of the reasons. Anyway I switched to using RxJavaHooks.setOnXXXScheduler

    RxJavaHooks.setOnComputationScheduler(computationScheduler);

static Func1<Scheduler, Scheduler> computationScheduler = new Func1<Scheduler, Scheduler>() {
    @Override
    public Scheduler call(Scheduler scheduler) {
      return convertToIdlingScheduler(scheduler);
    }
  };

 private static IdlingScheduler convertToIdlingScheduler(Scheduler scheduler) {
    IdlingScheduler idlingScheduler = new IdlingScheduler(scheduler);

    Espresso.registerIdlingResources(idlingScheduler.countingIdlingResource());

    return idlingScheduler;
  }

[Maragues]

@marksutt you are right, the increment() must be invoked outside of the call or there will be issues with delayed observables. The increment won't happen until the call is executed, so Espresso won't wait for that call to happen.

Also, I had to create static IdlingSchedulers for espresso to work properly on some tests. This is how my runner looks like now

private static IdlingScheduler computationIdlingScheduler, ioIdlingScheduler, newThreadIdlingScheduler;

  static Func1<Scheduler, Scheduler> computationScheduler = new Func1<Scheduler, Scheduler>() {
    @Override
    public Scheduler call(Scheduler scheduler) {
      if (computationIdlingScheduler == null)
        computationIdlingScheduler = convertToIdlingScheduler(scheduler);

      return computationIdlingScheduler;
    }
  };

  static Func1<Scheduler, Scheduler> ioScheduler = new Func1<Scheduler, Scheduler>() {
    @Override
    public Scheduler call(Scheduler scheduler) {
      if (ioIdlingScheduler == null)
        ioIdlingScheduler = convertToIdlingScheduler(scheduler);

      return ioIdlingScheduler;
    }
  };

  static Func1<Scheduler, Scheduler> newThreadScheduler = new Func1<Scheduler, Scheduler>() {
    @Override
    public Scheduler call(Scheduler scheduler) {
      if (newThreadIdlingScheduler == null)
        newThreadIdlingScheduler = convertToIdlingScheduler(scheduler);

      return newThreadIdlingScheduler;
    }
  };

[msgitter]

@Maragues thx for your feedback.

I think that the approach with counting actions is also not perfect. For example if some actions get scheduled but never executed, we timeout waiting for resource getting idle. I stumbled over this problem, when I was using the debounce operator. There may be other operators with similar issues.

[Maragues]

@marksutt I'm currently experiencing those issues with Observable.zip that uses Observable.interval as second source. A second TestRunner.schedulePeriodically never invokes its action, probably because the first one hasn't decremented the idlingResource yet.

I think I'm going to abandon this for now and wait for a better solution. @artem-zinnatullin mentioned [...] we decided to use condition watcher and "wait" for actual UI conditions (through idling API) instead of letting tests know how async operations work under the hood., perhaps that's a valid approach, tho I don't know how it's implemented.

[artem-zinnatullin]

@Maragues we use ConditionWatcher which allows you write conditions that will be checked periodically (pretty frequent) with timeout. Yes it's polling (not that CPU intensive, but ok to execute tests fast) instead of pushing but it allows us keep tests unaware of underlying async operations internal details and avoid problems you're facing with delayed actions.

Also, we have some background operations which may or may not affect flow under the test and there is no clear way to divide which of async things you need to wait to and which you don't with such global RxJava hooks or similar approaches.

[kmdupr33]

What's wrong with accomplishing this by subscribing on schedulers from the AsyncTask thread pool, which is already synced with espresso:

observable.subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))

[artem-zinnatullin]

It might change behavior of the app.

In real life you would probably use standard Schedulers.io() and computation(), first one is threads-unbounded and second one has threads count == CPU cores. AsyncTask.THREAD_POOL_EXECUTOR has different threads limit and it's work queue is limited by 128 tasks.

On Thu, Aug 11, 2016 at 3:40 PM, kmdupr33 notifications@github.com wrote:

What's wrong with accomplishing this by subscribing on schedulers from the AsyncTask thread pool, which is already synced with espresso:

observable.subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/ReactiveX/RxAndroid/issues/149#issuecomment-239149259, or mute the thread https://github.com/notifications/unsubscribe-auth/AA7B3DRqWzxch0kKQJEFn25FN-lbCoQoks5qexhRgaJpZM4DymfO .

[kmdupr33]

Good point. Just curious: have you run in to a lot of people for whom the AsyncTask work queue limit is actually problem in practice?

[realdadfish]

@marksutt / @cxzhang2's solution is still what I would prefer for now, minus that I completely skip "decoration" of delayed or periodically scheduled actions, which you would have to sync yourself. Lets see how long this proves being stable...

[digitalbuddha]

There seems to be bits and pieces in this PR that I'm having trouble wrapping my head around. It would be beneficial (to me at least) if someone could make a gist of what is needed.

[aahlenst]

We're intercepting on schedule and it's pretty solid so far (RxJava 2): https://gist.github.com/aahlenst/8d3958ceceb971465873e5818ba6fd1f

[nschwermann]

When you register the idle resources in the Scheduler hook like in Jake's example when do you unregister it?

[JakeWharton]

You don't

On Fri, Mar 10, 2017, 10:52 AM Nathan Schwermann notifications@github.com wrote:

When you register the idle resources in the Scheduler hook like in Jake's example when do you unregister it?

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/ReactiveX/RxAndroid/issues/149#issuecomment-285704913, or mute the thread https://github.com/notifications/unsubscribe-auth/AAEEEYS7-fIVuaaf5bQW8If5pwjsjCf3ks5rkXHFgaJpZM4DymfO .

[nschwermann]

Sadly the above proposal doesn't seem to work with Observable.timer + combineLatest

[tir38]

If I am willing to deal with the differences and I use observable.subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR)) then I should be able to put a breakpoint inside my subscription somewhere before observeOn() and call Espresso.getIdlingResources() to verify this is working.

someObservable
    .doOnNext((Action1<Order>) order -> {
        Log.d("TAG", "Espresso should have idling resource");
        // put breakpoint here and call Espresso.getIdlingResources()
    })
    subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(order -> {
        // TODO on next
    });

However the list of idling resources is empty.

[LuigiPapino]

I just modified a bit the solution proposed by @aahlenst I added a comment on his GIst, but I report the code also here.

Just modified a bit, because wasn't working for me. I don't use it with espresso for now, but in unit tests running with the AndroidJUnit.

the problem for me was, that the taskCount was increased in the returned Runnable, so the resource was busy only while executing the task. I wanted the resource to be busy from the instant the task is scheduled.

I also added an utility method that just wait until the resource is idle, so can be used in unit tests.

/*
 * The MIT License
 *
 * Copyright (c) 2016 Andreas Ahlenstorf
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

import android.support.test.espresso.IdlingResource;
import com.adaptics.dropkitchen.utils.Loggy;
import io.reactivex.functions.Function;
import io.reactivex.plugins.RxJavaPlugins;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class RxIdlingResource implements IdlingResource, Function<Runnable, Runnable> {

  private static final String TAG = RxIdlingResource.class.getSimpleName();

  private static final ReentrantReadWriteLock IDLING_STATE_LOCK = new ReentrantReadWriteLock();

  // Guarded by IDLING_STATE_LOCK
  private int taskCount = 0;

  // Guarded by IDLING_STATE_LOCK
  private ResourceCallback transitionCallback;

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

  @Override
  public boolean isIdleNow() {

    boolean result;

    IDLING_STATE_LOCK.readLock().lock();
    result = taskCount == 0;
    IDLING_STATE_LOCK.readLock().unlock();

    return result;
  }

  @Override
  public void registerIdleTransitionCallback(final ResourceCallback callback) {
    IDLING_STATE_LOCK.writeLock().lock();
    this.transitionCallback = callback;
    IDLING_STATE_LOCK.writeLock().unlock();
  }

  @Override
  public Runnable apply(final Runnable runnable) throws Exception {
    IDLING_STATE_LOCK.writeLock().lock();
    taskCount++;
    Log.d(TAG, "TaskCount increase " + taskCount);
    IDLING_STATE_LOCK.writeLock().unlock();
    return () -> {
      try {
        runnable.run();
      } finally {
        IDLING_STATE_LOCK.writeLock().lock();

        try {
          taskCount--;
          Log.d(TAG, "TaskCount decrease " + taskCount);

          if (taskCount == 0 && transitionCallback != null) {
            transitionCallback.onTransitionToIdle();
            Log.d(TAG, "idle ");
          }
        } finally {
          IDLING_STATE_LOCK.writeLock().unlock();
        }
      }
    };
  }

  public void waitForIdle() {
    if (!isIdleNow()) {
      CountDownLatch latch = new CountDownLatch(1);
      registerIdleTransitionCallback(latch::countDown);
      try {
        latch.await();
      } catch (InterruptedException e) {
        Log.e(TAG, e.getMessage(), e);
      }
    }
  }

  public void register() {
    RxJavaPlugins.setScheduleHandler(this);
  }
}

Just init in the setup

 RxIdlingResource rxIdlingResource = new RxIdlingResource();
  @Before
  public void setUp() throws Exception {
    rxIdlingResource = new RxIdlingResource();
    rxIdlingResource.register();
  }

And use it in the test:

@Test
public void testCool(){

   usecase.executeFetchAndSave(); //background task with retrofit and rxjava
   rxIdlingResource.waitForIdle();

  //your assertions
}

[viniciussoares]

A simpler version:

public class RxEspressoScheduleHandler implements Function<Runnable, Runnable> {

  private final CountingIdlingResource countingIdlingResource = 
      new CountingIdlingResource("rxJava");

  @Override
  public Runnable apply(@NonNull final Runnable runnable) throws Exception {
    return new Runnable() {
      @Override
      public void run() {
        countingIdlingResource.increment();

        try {
          runnable.run();
        } finally {
          countingIdlingResource.decrement();
        }
      }
    };
  }

  public IdlingResource getIdlingResource() {
    return countingIdlingResource;
  }

}

Usage:

RxEspressoScheduleHandler rxEspressoScheduleHandler = new RxEspressoScheduleHandler();
RxJavaPlugins.setScheduleHandler(rxEspressoScheduleHandler);
Espresso.registerIdlingResources(rxEspressoScheduleHandler.getIdlingResource());

[handstandsam]

There is also a library for this here: https://github.com/square/RxIdler