RxJava的真實世界用例對象

Question

有很多很好的文章解釋關於RxJava。但幾乎沒有一個用現實世界的例子來解釋這個概念。

所以我基本上明白RxJava主題像管的概念，它既是observable和observer。

但我不明白這個RxJava主題在Android開發世界中的真實用法。你能詳細說明一下嗎？

Answer 1

在我的情況是因爲我有一個Observable正在等待另一個Observable的項目，因爲它是Interval，所以這是異常的。

Scheduler scheduler = RxHelper.scheduler(vertx.getOrCreateContext()); 


Observable.just(callAnotherObservable) 
      .subscribe(item -> System.out.println(item) 

public Observable<String> callAnotherObservable(Scheduler scheduler,){ 
      Subject subject = ReplaySubject.create(1); 
      Observable.interval(100,TimeUnit.MILLISECONDS) 
      .map(i->"item to be passed to the other observable") 
      .subscribe(subject); 
      return subject.observeOn(scheduler).first();//Here we wait for the first emission of the interval Observable. 

} 

在這裏，你可以看到我們使用subject.first（）等待的時間間隔觀察到的第一排放，這在另一個線程中運行。

如果你想看到「hotObservables」 https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/connectable/HotObservable.java

Answer 2

主題更多的例子有很多的「真實世界」的應用特別是當你逐漸從必須改變你的代碼庫，以無功風格。它可以作爲這兩個世界之間的橋樑，在這兩個世界中，您可以通過流之外的非反應式代碼影響流。

但正如你所要求的一個例子。最近我在用戶嘗試從活動返回時實現自定義行爲。 RxJava爲我面臨的問題提供了非常優雅的解決方案，因此我需要編寫一系列與用戶想要返回的事件相對應的事件流。我特意避開了「按下後退按鈕」一詞，因爲代碼庫中有幾個地方可以模擬返回的情況，並且它總是通過onBackPressed()方法。

將此轉換爲單個流將需要進行大規模的重構，目前這不是一個預算。但我不想通過RxJava放棄解決方案，因爲它可以使代碼更加簡潔。使用BehaviorSubject給出了答案，因爲我只需要在onBackPressed()方法中發出事件。

Answer 3

我爲Android應用程序構建了一個通用的ReportDownloadManager，我們需要使用一個Observable，它的觀察者可以在本地下載和存儲一個文件。成功或失敗的下載事件需要經理處理，但也需要將Observable公開給使用此DownloadManager的Activities/Services。我認爲這是一個很好的用例，它使用一個Subject來消費最初的Observable，但也爲客戶端Observable生成事件。

import android.app.DownloadManager; 
import android.content.Context; 
import android.webkit.MimeTypeMap; 

import java.io.File; 

import io.reactivex.Observable; 
import io.reactivex.functions.Consumer; 
import io.reactivex.subjects.PublishSubject; 

public class ReportDownloadManager { 

    private final DownloadManager platformDownloadManager; 

    public ReportDownloadManager(Context context) { 
     this.platformDownloadManager = (DownloadManager) context.getSystemService(Context.DOWNLOAD_SERVICE); 
    } 

    public Observable<Object> download(final File file, DownloadAction downloadAction) { 
     final PublishSubject<Object> subject = PublishSubject.create(); 
     downloadAction.execute(file.getName()) 
       .subscribe(new Consumer<Object>() { 
        @Override 
        public void accept(Object o) throws Exception { 
         platformDownloadManager.addCompletedDownload(file.getName(), "No description", false, 
           MimeTypeMap.getSingleton().getMimeTypeFromExtension("pdf"), file.getAbsolutePath(), 
           file.length(), true); 
         subject.onNext(new Object()); 
         subject.onComplete(); 
        } 
       }, new Consumer<Throwable>() { 
        @Override 
        public void accept(Throwable throwable) throws Exception { 
         subject.onError(throwable); 
        } 
       }); 
     return subject; 
    } 

    interface DownloadAction { 
     Observable<Object> execute(String fileAbsolutePath); 
    } 

}