Làm thế nào để kiểm tra rằng có thể quan sát sử dụng đúng lịch trong RxJava?

Question

Tôi có lớp sau (giản thể):

public class Usecase<T> { 
    private final Observable<T> get; 
    private final Scheduler observeScheduler; 

    public Usecase(Observable<T> get, Scheduler observeScheduler) { 
     this.get = get; 
     this.observeScheduler = observeScheduler; 
    } 

    public Observable<T> execute() { 
     return get.subscribeOn(Schedulers.io()).observeOn(observeScheduler); 
    } 
} 

Và tôi đang viết các unit test cho nó. Làm cách nào để kiểm tra rằng subscribeOn và observeOn được gọi với giá trị chính xác?

tôi thử như sau:

Observable<String> observable = mock(Observable.class); 
    Usecase<String> usecase = new Usecase(observable, Schedulers.computation()); 
    usecase.execute(); 

    verify(observable).subscribeOn(Schedulers.computation()); // should fail here, but passes 
    verify(observable).observeOn(Schedulers.computation()); // should pass, but fails: Missing method call for verify(mock) here 

ở trên không (tôi nghĩ) vì subscribeOn và observeOn là final phương pháp. Vì vậy, có thể có một số cách khác để đảm bảo rằng các quan sát sử dụng đúng lịch?

Answer 1

There is a way để truy cập gián tiếp cả hai chủ đề mà một quan sát đang hoạt động và được quan sát, có nghĩa là bạn có thể xác minh rằng Observable sử dụng đúng lịch biểu.

Chúng tôi bị giới hạn xác minh chủ đề theo tên. May mắn thay, các chủ đề được sử dụng bởi Schedulers.io() được đặt tên với một tiền tố nhất quán mà chúng ta có thể đối sánh. Dưới đây là (full?) Danh sách các schedulers có tiền tố độc đáo để tham khảo:

• Schedulers.io() - RxCachedThreadScheduler
• Schedulers.newThread()-RxNewThreadScheduler
• Schedulers.computation()-RxComputationThreadPool

Để xác minh một Đài quan sát là đăng ký để trên thread IO:

// Calling Thread.currentThread() inside Observer.OnSubscribe will return the 
// thread the Observable is running on (Schedulers.io() in our case) 
Observable<String> obs = Observable.create((Subscriber<? super String> s) -> { 
    s.onNext(Thread.currentThread().getName()); 
    s.onCompleted(); 
}) 

// Schedule the Observable 
Usecase usecase = new Usecase(obs, Schedulers.immediate()); 
Observable usecaseObservable = usecase.execute(); 

// Verify the Observable emitted the name of the IO thread 
String subscribingThread = usecaseObservable.toBlocking().first(); 
assertThat(subscribingThread).startsWith("RxCachedThreadScheduler"); 

Để xác minh một Quan sát là quan sát trên thread tính toán, bạn có thể sử dụng để truy cập TestSubscriber#getLastSeenThread thread mới được sử dụng để quan sát.

TestSubscriber<Object> subscriber = TestSubscriber.create(); 
UseCase usecase = new UseCase(Observable.empty(), Schedulers.computation()) 
usecase.execute().subscribe(subscriber); 

// The observable runs asynchronously, so wait for it to complete 
subscriber.awaitTerminalEvent(); 
subscriber.assertNoErrors(); 

// Verify the observable was observed on the computation thread 
String observingThread = subscriber.getLastSeenThread().getName(); 
assertThat(observingThread).startsWith("RxComputationThreadPool"); 

Không thư viện của bên thứ ba hoặc nhạo báng là cần thiết, mặc dù tôi đang sử dụng AssertJ cho thông thạo startsWith khẳng định.

Answer 2

Áp dụng toán tử trên Observable trả về Observable mới do đó bất kỳ ứng dụng toán tử nào sau đó sẽ xảy ra trên một đối tượng khác. Bạn sẽ phải theo biểu đồ thành phần trong một Observable để khám phá những toán tử nào đã được áp dụng và với các tham số nào.

Điều này không được hỗ trợ trong RxJava và bạn phải dựa vào các chi tiết và phản ánh nội bộ.

Nói chung, bạn không thể giả định bất cứ điều gì về vị trí nơi sự kiện sẽ đến từ bạn, nhưng bạn có thể áp dụng observeOn để đảm bảo họ truy cập vào chuỗi của bạn.