如何在调用rx.Observable.sample（）时接收最后的序列发射？

Question

我正在从远程URL读取文件并使用RxJava报告下载进度。文件编写器Observable发出一系列DownloadProgress对象。由于很多项目正在发射，我使用Observable.sample（）来管理背压。这非常有效 - UI更新以恒定的速率进行，并且没有背压问题，但最后的进度更新几乎总是被忽略。

我想收到Observable序列中的最后一项，以便我可以用最终进度更新UI。确保始终发出Observable序列中最后一项的最佳方法是什么？

Observable<Response> fileReader = 
     Rx.okHttpGetRequest(url); 
OkHttpResponseWriter fileWriter = 
     Rx.okHttpResponseWriter(outFile); 

Subscription subscription = fileReader.flatMap(fileWriter) 
     .sample(1, TimeUnit.SECONDS) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Subscriber<DownloadProgress>() { 
      @Override 
      public void onCompleted() {} 
      @Override 
      public void onError(Throwable e) {} 
      @Override 
      public void onNext(DownloadProgress progress) { 
       // I want to receive an update every one second 
       // I also want to always receive the last progress update 
      } 
     }); 

Answer 1

另外，使用buffer代替sample，将让你发射的“额外“最后的人。我建议这样做，这样你就不会失去任何可组合性，也不会为将来的开发人员（包括未来的你）带来麻烦。

private Subscriber<List<Integer>> loggingSubscriber2 = new SimpleSubscriber<List<Integer>>() { 
    @Override 
    public void onNext(List<Integer> integers) { 
     Log.v(TAG, String.valueOf(integers.get(integers.size() - 1))); 
    } 
}; 

private void startObservableTests() { 
    Observable<Integer> fileObserver = Observable.create(
      new Observable.OnSubscribe<Integer>() { 
       @Override 
       public void call(Subscriber<? super Integer> subscriber) { 
        for (int i = 1; i <= 9; i++) { 
         subscriber.onNext(i); 
         try { 
          Thread.sleep(100); 
         } catch (InterruptedException e) { 
          e.printStackTrace(); 
         } 
        } 
        subscriber.onCompleted(); 
       } 
      }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()); 

    fileObserver.buffer(500, TimeUnit.MILLISECONDS).subscribe(loggingSubscriber2); 
} 

Answer 2

这是我结束了：

private void startObservableTests() { 

    Observable<Integer> fileObserver = Observable.create(
      new Observable.OnSubscribe<Integer>() { 
       @Override 
       public void call(Subscriber<? super Integer> subscriber) { 
        for (int i = 1; i <= 9; i++) { 
         subscriber.onNext(i); 
         try { 
          Thread.sleep(100); 
         } catch (InterruptedException e) { 
          e.printStackTrace(); 
         } 
        } 
        subscriber.onCompleted(); 
       } 
      }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()); 


    fileObserver.sample(500, TimeUnit.MILLISECONDS).subscribe(new SuperSubscriber(fileObserver)); 
} 

private class SuperSubscriber extends Subscriber<Integer> { 

    Observable obs; 

    public SuperSubscriber(Observable<Integer> fileObserver) { 
     obs = fileObserver; 
    } 

    @Override 
    public void onCompleted() { 
     obs.last().subscribe(new SimpleSubscriber<Integer>() { 
      @Override 
      public void onNext(Integer o) { 
       Log.v(TAG, "final value was " + o); 
      } 
     }); 
    } 

    @Override 
    public void onError(Throwable e) { 

    } 

    @Override 
    public void onNext(Integer o) { 
     Log.v(TAG, "got a " + o); 
    } 
} 

这是输出：

10-01 15:18:36.893 V/TAG (26129): got a 5 
10-01 15:18:38.214 V/TAG (26129): final value was 9 

Answer 3

以@Travis发布的优秀解决方案为基础，以下是我最终的代码。代替buffer，我组合使用window与switchOnNext做一些进一步的处理：

Observable<Response> fileReader = 
    Rx.okHttpGetRequest(fileInfo.getUrl()); 
OkHttpResponseWriter fileWriter = 
    Rx.okHttpResponseWriter(outFile, totalBytesRead); 
Subscription subscription = Observable.switchOnNext(fileReader 
     .flatMap(fileWriter) 
     .window(1, TimeUnit.SECONDS)) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Subscriber<DownloadProgress>() { 
      @Override 
      public void onCompleted() {} 
      @Override 
      public void onError(Throwable e) {} 
      @Override 
      public void onNext(DownloadProgress progress) { 
       // always emits the most recent DownloadProgress, even the last one! 
      } 
}); 

说明：

• fileReader需要URL，创建一个HTTP请求并发出Response
• fileReader.flatMap(fileWriter)取Response并将字节流写入磁盘，发出一系列DownloadProgress对象
• window(1, TimeUnit.Seconds)发出List<DownloadProgress>对象每一秒，然后包装并重新发出他们为Observable<DownloadProgress>
• Observable.switchOnNext()注意到最近发出Observable<DownloadProgress>并发出最后DownloadProgress对象序列中