RxJava 2: Observable.create() warte, bis der Async-Code mit Thread.sleep() fertig ist

Question

Ich wickle etwas Async-Code mit Observable.create(...). Der Async-Code muss fünf Sekunden lang "aktiv" sein und dann sollte er gestoppt werden. Dies ist, wie ich zur Zeit bin es zu stoppen:

Observable.<MyObject>create(emitter -> { 

    // Some async code 

    emitter.setDisposable(Disposables.fromRunnable(() -> { 
     // Stop the above async code 
    })); 

    // Wait for 5 seconds until the async code above has had enough time to finish its task. 
    Thread.sleep(5000); 

    if (!emitter.isDisposed()) { 
     // Stop the above async code 
     emitter.setDisposable(null); 
    } 
})... 

Ich bin nicht sicher, ob dies der beste Weg ist, weil Thread.sleep(5000) verwendet wird. Gibt es einen besseren Weg, dies mit etwas RxJava-Code zu tun?

Answer 1

Es gibt eine zeitliche Überlastung von take, die Sie Elemente aus einer Quelle, bis die vorgegebene Zeit abgelaufen haben können:

Observable.<MyObject>create(emitter -> { 

    // Some async code 

    emitter.setDisposable(Disposables.fromRunnable(() -> { 
     // Stop the above async code 
    })); 
}) 
.take(5, TimeUnit.SECONDS); 

Sie noch haben die Einrichtung des Disposable in create.