如何從Runnable創建Observable？

Question

有時我想觸發Runnable作爲我的Observable序列的一部分，但Runnable不報告進度。

我寫了一個簡單的工廠用於包裝Runnable對象爲Observable：

public static <T> Observable<T> fromRunnable(final Runnable action) { 
    if (action == null) { 
     throw new NullPointerException("action"); 
    } 
    return Observable.fromPublisher(subscriber -> { 
     try { 
      action.run(); 
      subscriber.onComplete(); 
     } catch (final Throwable throwable) { 
      subscriber.onError(throwable); 
     } 
    }); 
} 

用法：

Observable.concat(
    someTask, 
    MoreObservables.fromRunnable(() -> { 
     System.out.println("Done. "); 
    })); 

但是否RxJava 2已經提供了這個功能？

Answer 1

Observable,但是Completable可以由Runnable製成沒有這樣的工廠方法。所以，你可以創建一個Completable第一，然後將其轉換爲Observable：

Observable.concat(
    someTask, 
    Completable.fromRunnable(() -> { 
     System.out.println("Done"); 
    }).toObservable() 
); 

---

更新：有例外

Completable.fromRunnable內部捕獲來自其Runnable異常並將其推入流作爲onError處理排放。但是，如果您使用的是Java，則必須自己處理run()方法中的已檢查異常。爲了避免你可以利用Callable而不是Runnable，因爲它的call()方法的簽名聲明它可以拋出異常。 Completable.fromCallable()包裝紙異常轉換成onError排放以及：

Observable.concat(
    someTask, 
    Completable.fromCallable(() -> { 
     System.out.println("Done"); 
     return null; 
    }).toObservable() 
); 

另外Callable可用於與單個項發射創建Observable或Single。

P.S.查看源代碼，這些方法非常簡單。

P.P.S.Kotlin還沒有檢查異常;）

---

更新2

還有fromAction用於創建Completable工廠方法。它接受Action對象。

一個類似於Runnable的函數接口，但允許拋出一個檢查的異常。

因此，代碼可以簡化爲：

Observable.concat(
    someTask, 
    Completable.fromAction(() -> { 
     System.out.println("Done"); 
    }).toObservable() 
);