- Learning RxJava
- Thomas Nield
- 159字
- 2021-07-02 22:22:55
Handling Disposal with Observable.create()
If your Observable.create() is returning a long-running or infinite Observable, you should ideally check the isDisposed() method of ObservableEmitter regularly, to see whether you should keep sending emissions. This prevents unnecessary work from being done if the subscription is no longer active.
In this case, you should use Observable.range(), but for the sake of the example, let's say we are emitting integers in a for loop in Observable.create(). Before emitting each integer, you should make sure that ObservableEmitter does not indicate that a disposal was called:
import io.reactivex.Observable;
public class Launcher {
public static void main(String[] args) {
Observable<Integer> source =
Observable.create(observableEmitter -> {
try {
for (int i = 0; i < 1000; i++) {
while (!observableEmitter.isDisposed()) {
observableEmitter.onNext(i);
}
if (observableEmitter.isDisposed())
return;
}
observableEmitter.onComplete();
} catch (Throwable e) {
observableEmitter.onError(e);
}
});
}
}
If your Observable.create() is wrapped around some resource, you should also handle the disposal of that resource to prevent leaks. ObservableEmitter has the setCancellable() and setDisposable() methods for that. In our earlier JavaFX example, we should remove the ChangeListener from our JavaFX ObservableValue when a disposal occurs. We can provide a lambda to setCancellable(), which will execute the following action for us, which will occur when dispose() is called:
private static <T> Observable<T> valuesOf(final ObservableValue<T>
fxObservable) {
return Observable.create(observableEmitter -> {
//emit initial state
observableEmitter.onNext(fxObservable.getValue());
//emit value changes uses a listener
final ChangeListener<T> listener =
(observableValue, prev, current) ->
observableEmitter.onNext(current);
//add listener to ObservableValue
fxObservable.addListener(listener);
//Handle disposing by specifying cancellable
observableEmitter.setCancellable(() ->
fxObservable.removeListener(listener));
});
}
- 大話PLC(輕松動(dòng)漫版)
- 零基礎(chǔ)學(xué)Scratch少兒編程:小學(xué)課本中的Scratch創(chuàng)意編程
- Python神經(jīng)網(wǎng)絡(luò)項(xiàng)目實(shí)戰(zhàn)
- OpenShift在企業(yè)中的實(shí)踐:PaaS DevOps微服務(wù)(第2版)
- Building Machine Learning Systems with Python(Second Edition)
- Qlik Sense? Cookbook
- Python開(kāi)發(fā)基礎(chǔ)
- Mudbox 2013 Cookbook
- Python全棧開(kāi)發(fā):基礎(chǔ)入門
- Deep Learning for Natural Language Processing
- Java多線程并發(fā)體系實(shí)戰(zhàn)(微課視頻版)
- PostgreSQL 12 High Availability Cookbook
- Spring Web Services 2 Cookbook
- MonoTouch應(yīng)用開(kāi)發(fā)實(shí)踐指南:使用C#和.NET開(kāi)發(fā)iOS應(yīng)用
- SQL Server 2008數(shù)據(jù)庫(kù)應(yīng)用技術(shù)(第2版)