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“故障排除”，“竞争条件”中的更多说明。