RxJava đăng ký và quan sát trên cùng một luồng như kiểm tra đơn vị

Question

Tôi muốn viết một loại "kiểm tra hộp đen" cho một thành phần, sử dụng nội bộ RxJava.

Nội bộ nó sử dụng Retrofit trả về Observable để tạo httpcall và sau đó sử dụng .flatmap() để xử lý trong tương lai dữ liệu được truy xuất từ ​​trang bị thêm. Ý tưởng là để cung cấp cho rằng thành phần một Transformer cho thiết lập các schedulers vào người quan sát như thế này:

class DefaultTransformer <T> implements Transformer<T, T> { 

    public Observable<T> call(Observable<T> observable) { 
     return observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()); 
    } 
} 

Component tôi làm điều gì đó như thế này:

void execute(Transformer<T, T> scheduler){ 
    Observable<List<Team>> observable = retrofitApi.getLeague(leagueId, seasonId) 
     .flatMap(new Func1<LeagueWrapper, Observable<List<Team>>>() { 
      @Override public Observable<List<Team>> call(LeagueWrapper wrapper) {     
      return Observable.just(wrapper.getLeague().getTeams()); 
      } 
     }); 

    observable.compose(transformer); 

    observable.subscribe(this); 
} 

Trong sản xuất tôi vượt qua DefaultTransformer như tham số, nhưng đối với Bài kiểm tra đơn vị tôi muốn gửi Transformer chạy trên cùng một luồng với bài kiểm tra đơn vị, vì vậy mọi thứ sẽ chạy đồng bộ (không đồng bộ).

Tôi cố gắng đó:

class UnitTestTransformer <T> implements Transformer<T, T> { 

     public Observable<T> call(Observable<T> observable) { 
      return observable.subscribeOn(Schedulers.test()).observeOn(AndroidSchedulers.test()); 
     } 
    } 

Nhưng nó vẫn chạy async trong các thử nghiệm đơn vị của tôi. Tôi cũng đã thử Scheduler.immediate(). toBlocking() có vẻ không phải là một tùy chọn, bởi vì nó không còn là Observable nữa. Bất cứ ý tưởng những gì có thể là sai?

Answer 1

Nếu thay đổi kiểu cách cách execute() được gọi không phải là một tùy chọn, bạn có thể muốn thử sử dụng cơ chế Plugin RxJava.

https://github.com/ReactiveX/RxJava/wiki/Plugins

Bạn có thể cung cấp:

• RxJavaSchedulersHook để ghi đè lên schedulers được cung cấp trong quá trình thực hiện kiểm tra và nhận được chúng để thực hiện đồng bộ
• RxJavaObservableExecutionHook để móc vào các đường ống dẫn thực hiện Quan sát và sử dụng một số loại phương thức đồng bộ hóa (như số CountdownLatch) để đợi đăng ký Quan sát được hoàn thành trước khi tiếp tục

Answer 2

Tôi đã có một vấn đề tương tự và giải quyết nó bằng cách sử dụng các TestObserver lớp (http://reactivex.io/RxJava/javadoc/rx/observers/TestObserver.html)

Nó có ba phương pháp sau đây cho phép truy cập đến các sự kiện nhận:

getOnCompletedEvents() 
getOnErrorEvents() 
getOnNextEvents() 

Tôi hy vọng điều này có thể giúp bạn cũng vậy, nếu bạn có thể tiêm thuê bao của bạn bằng cách nào đó. Dưới đây là ví dụ về cách tôi đã thử nghiệm:

TestObserver<MyModel> testObserver = new TestObserver<>(); 
myObservableSupplyingMethod().subscribe(testObserver); 

assertThat(testObserver.getOnErrorEvents().size()).isEqualTo(0); 
assertThat(testObserver.getOnNextEvents().get(0)).isNotNull(); 
...