RxJava Unittests

Question

我無法測試我的RxJava代碼，因爲它似乎總是運行測試而不等待調用subscribe（）時應該發生的執行。

這是存儲庫進行測試：

public class Repository<T extends Entity> { 

    private final DataSource<T> remoteDataSource; 
    private final DataSource<T> localDataSource; 

    public Repository(DataSource<T> remoteDataSource, DataSource<T> localDataSource) { 
     this.remoteDataSource = remoteDataSource; 
     this.localDataSource = localDataSource; 
    } 

    public Observable<Boolean> save(T entity) { 

     return localDataSource 
       .save(entity) 
       .flatMap(success -> { 
        if (success) { 
         return remoteDataSource.save(entity); 
        } 
        return Observable.just(Boolean.FALSE); 
       }) 
       .doOnNext(success -> { 
        if (success) { 
         if (cache == null) { 
          cache = new LinkedHashMap<>(); 
         } 
         cache.put(entity.getId(), entity); 
        } 
       }); 
    } 
} 

此DataSource接口：

public interface DataSource<T extends Entity> { 
    <T> Observable<T> get(); 

    Observable<Boolean> save(T entity); 

    Observable<Boolean> clear(); 

    Observable<Boolean> remove(T entity); 
} 

而這是單元測試包括MockDataSource即應該以模擬不同的執行數據訪問計時：

public class RepositoryTest { 

    @Test 
    public void testRepository() { 

     MockSource remoteSource = new MockSource("RemoteSource", 1000L); 
     MockSource localSource = new MockSource("LocalSource", 200L); 
     Repository<Poi> poiRepository = new Repository<>(remoteSource, localSource); 

     Poi poi1 = newMockPoi(); 

     Observable<Boolean> obs = poiRepository.save(poi1); 
     TestSubscriber<Boolean> testSubscriber = new TestSubscriber<>(); 
     obs.subscribe(testSubscriber); 

     testSubscriber.assertNoErrors(); 
     testSubscriber.assertReceivedOnNext(Arrays.asList(true)); 


    } 

    private Poi newMockPoi() { 
     Poi poi = new Poi(); 
     poi.name = RandomStringUtils.randomAlphabetic(12); 
     poi.description = RandomStringUtils.randomAlphabetic(255); 
     poi.latitude = new Random().nextDouble(); 
     poi.longitude = new Random().nextDouble(); 
     return poi; 
    } 


    private class Poi extends Entity { 
     String name; 
     String description; 
     Double latitude; 
     Double longitude; 
    } 

    private class MockSource implements DataSource<Poi> { 

     private String name; 
     private final long delayInMilliseconds; 

     private Map<Long, Poi> pois = new LinkedHashMap<>(); 

     private MockSource(String name, long delayInMilliseconds) { 
      this.delayInMilliseconds = delayInMilliseconds; 
      this.name = name; 
     } 

     @Override 
     public Observable<List<Poi>> get() { 
      return Observable 
        .zip(
          Observable 
            .just(pois) 
            .map(Map::entrySet) 
            .flatMapIterable(entries -> entries) 
            .map(Map.Entry::getValue) 
            .toList(), 
          Observable 
            .interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs) 
        .doOnNext(pois -> System.out.println("Soure " + name + " emitted entity")); 
     } 

     @Override 
     public Observable<Boolean> save(Poi entity) { 
      return Observable 
        .zip(
          Observable.just(true).asObservable(), 
          Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs) 
        .doOnNext(value -> pois.put(entity.getId(), entity)) 
        .doOnNext(pois -> System.out.println("Soure " + name + " saved entity")); 
     } 

     @Override 
     public Observable<Boolean> clear() { 
      return Observable 
        .zip(
          Observable.just(true).asObservable(), 
          Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs) 
        .doOnNext(value -> pois.clear()) 
        .doOnNext(pois -> System.out.println("Soure " + name + " cleared all entities")); 
     } 

     @Override 
     public Observable<Boolean> remove(Poi entity) { 
      return Observable 
        .zip(
          Observable.just(true).asObservable(), 
          Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs) 
        .doOnNext(value -> pois.remove(entity)) 
        .doOnNext(pois -> System.out.println("Soure " + name + " removed entity")); 
     } 
    } 
} 

這是輸出：

java.lang.AssertionError: Number of items does not match. Provided: 1 
Actual: 0. 
Provided values: [true] 
Actual values: [] 
(0 completions) 

at rx.observers.TestSubscriber.assertionError(TestSubscriber.java:667) 
at rx.observers.TestSubscriber.assertReceivedOnNext(TestSubscriber.java:320) 
at nl.itc.geofara.app.data.source.RepositoryTest.testRepository(RepositoryTest.java:37) 

此外，在存儲庫的保存方法中設置斷點並在調試模式下運行時，將顯示例如.flatMap（成功 - > ...）永遠不會被調用。

Answer 1

您遇到競爭條件可觀察的執行和JUnit線程之間，因爲您在模擬源使用Observable.interval介紹implicit subscription on computation scheduler：

調度： 間隔默認的工作計算調度器。

此外，請參閱this answer「故障排除」，「競爭條件」中的更多說明。