Wie zu testen, ob Observable die richtigen Scheduler in RxJava verwendet?

Question

Ich habe die folgende Klasse (vereinfacht):

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); 
    } 
} 

Und ich schreibe Unit-Tests für sie. Wie kann ich testen, dass subscribeOn und observeOn mit korrekten Werten aufgerufen wurden?

Ich versuche folgendes:

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 

Die oben nicht (glaube ich), weil subscribeOn und observeOn sind final Methoden. Kann es also einen anderen Weg geben, um sicherzustellen, dass das Observable korrekte Scheduler verwendet?

Answer 1

There is a way indirekt Zugriff auf beide Threads, auf denen eine beobachtbare Funktion ausgeführt wird und beobachtet wird, was bedeutet, dass Sie tatsächlich überprüfen können, Observable verwendet die richtigen Scheduler.

Wir beschränken uns auf die Überprüfung von Threads nach Namen. Glücklicherweise werden Threads, die von Schedulers.io() verwendet werden, mit einem konsistenten Präfix benannt, mit dem wir übereinstimmen können. Hier ist die (voll?) Liste der Disponenten, die eindeutige Präfixe als Referenz haben:

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

zu überprüfen Sie eine Observable war abonniert zu auf dem IO Thread:

// 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"); 

beobachtbares Um zu überprüfen, war auf der Berechnung beobachtet Thread Sie TestSubscriber#getLastSeenThread verwenden, um den letzten Faden zur Beobachtung verwendet zuzugreifen.

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"); 

Keine Bibliotheken von Drittanbietern oder spöttische sind notwendig, obwohl ich AssertJ für die fließend startsWith Behauptung verwenden.

Answer 2

Das Anwenden von Operatoren auf die Observable gibt eine neue Observable zurück, so dass jede nachfolgende Operatoranwendung auf einem anderen Objekt ausgeführt wird. Sie müssten dem Zusammensetzungsdiagramm in einem Observable folgen, um herauszufinden, welche Operatoren angewendet wurden und mit welchen Parametern.

Dies wird in RxJava nicht unterstützt und Sie müssen sich auf interne Details und Überlegungen verlassen.

Im Allgemeinen können Sie nicht von dem Ort ausgehen, an dem die Ereignisse kommen, aber Sie können observeOn anwenden, um sicherzustellen, dass sie zum Thread Ihrer Wahl gelangen.