UserRegionalCache { public Observable<List<UserSearchByLocationResponse.User>> searchByLocation( BoundingBox boundingBox) { return mAppScopeUserRegionalCache.searchByLocation(boundingBox, mUserRepository); } @Inject  UserRegionalCache(); Collection<UserSearchByLocationResponse.User> getAllCached(); Observable<List<UserSearchByLocationResponse.User>> searchByLocation(
            BoundingBox boundingBox);  }
@Test public void testSearchByLocationTimeoutOnDisposed() { WSTest.UncaughtRxExceptionHelper uncaughtRxExceptionHelper = WSTest.createUncaughtRxExceptionHelper(); final Lock lock = new ReentrantLock(); final AtomicBoolean requestStarted = new AtomicBoolean(false); final AtomicBoolean requestServed = new AtomicBoolean(false); mMockWebServer.setDispatcher(new Dispatcher() { @Override public MockResponse dispatch(RecordedRequest request) throws InterruptedException { if (!request.getPath().equals("/services/rest/hosts/by_location")) { return new MockResponse() .setStatus("HTTP/1.1 500") .setBody("[\"Unhandled\"]"); } requestStarted.set(true); try { lock.lock(); return new MockResponse() .setStatus("HTTP/1.1 500") .setBody("[\"Mock error\"]"); } finally { requestServed.set(true); lock.unlock(); } } }); try { lock.lock(); Disposable disposable = mUserRegionalCache.searchByLocation(new BoundingBox(0, 3, 2, 1)) .subscribe( searchResult -> fail("No result expected from disposed"), throwable -> fail("No error expected from disposed")); for (int i = 0; i < 10; i++) { if (requestStarted.get()) { break; } WSTest.sleepAndFlush(100); } assertThat(requestStarted.get()).isTrue(); disposable.dispose(); } finally { lock.unlock(); } for (int i = 0; i < 10; ++i) { WSTest.sleepAndFlush(100); } assertThat(requestServed.get()).isTrue(); uncaughtRxExceptionHelper.assertNoException(); }