ストリームを変更しない副作用サブスクリプションを使用するにはどうすればよいですか？

Question

シナリオ：

私はデータリスト（ビジネスの取引）を持っています。次の2つの条件が満たされた場合、このデータは無効と見なされます。

• ユーザー（UI /ビュー）をデータソースに加入している（rx.Observable<List<Transaction>>）
• > = 5分、最後のネットワーク同期経ちました。この値はシリアライズされ、データベースから観察されます。

注：UI /ビューがデータソースに加入されていない、と> = 5分が経過している場合、私は（誰も聞いていないので）

一部を新鮮なデータを同期する必要はありません以下のコードサンプルの詳細：

• data.observe()最後の5分で同期があった場合には戻りrx.Observable<List<Transaction>>
• void syncIfLast5Minutes()は命令的にチェックします - いない場合は、新しいネットワーク要求を取得するために実行されます新鮮なデータが

今、私は簡単のような何か行うことができます：

data.observe() 
    .doOnSubscribe(transactions -> syncIfLast5Minutes() 

をしかし、データがUIによる初期サブスクリプションに古くなっている場合にのみチェックします。 5分後にUIが引き続き購読されている場合、自動リフレッシュはここでトリガーされません（ただし、必要です）。データが古くなっているとき、私は

• は、元のストリームには影響しません副作用演算子を探しています

  は

• サブスクライブは、監視するタイマーを起動し（> = 5分が経過している）
• unsusbscribingは反応的にこれを達成するためにRxJavaで慣用の方法があります。このタイマー

キャンセル？

Answer 1

ストリームをObservable.intervalにマージして、副作用（データの更新）を実行し、ignoreElementsが適用されています。

これは実施例である：

import java.util.concurrent.TimeUnit; 

import rx.Observable; 
import rx.subjects.PublishSubject; 

public class Main { 

    public static void main(String[] args) throws InterruptedException { 
     PublishSubject<String> data = PublishSubject.create(); 
     //refresh data every second for this demo 
     //for your use case it's every 5 minutes 
     data.mergeWith(Observable.interval(1, TimeUnit.SECONDS) 
       // cause timed side-effect that updates data stream 
       .doOnNext(n -> data.onNext(n + "")) 
       .ignoreElements() 
       .cast(String.class)) 
      .doOnNext(System.out::println) 
      .toBlocking() 
      .subscribe(); 
    } 

} 

複数の加入者は、あなたが.shareで見たいと思うかもしれませんストリームを使用している場合はその一つだけリフレッシュ動作が5分ごとに発生します。