RX Observableを動的に更新する方法は？

Question

（RxKotlinとRxJavaでの作業が、簡単のためにメタコードを使用して）

多くReactive Extensionsガイドは、既に利用可能なデータからObservableを作成することから始めます。 The introduction to Reactive Programming you've been missingから、今outStream生み出す単純なフィルタを、考える人口リスト

val list = listOf(1,2,3,4,5) 
list.toObservable()  

から、RxKotlinのフロントページから、

var soureStream= Rx.Observable.just('https://api.github.com/users'); 

同様に、単一の文字列から作成されています

var outStream = sourceStream.filter({x > 3}) 

両方のガイドで、ソースイベントはaprioriと宣言されています。イベントのタイムラインを意味し、何らかの形

source: ----1,2,3,4,5------- 
out: --------------4,5--- 

はどうすればパイプラインのよりになるためにsourceStreamを変更することができたのか？つまり、sourceStreamの作成中に入力データがありませんか？ソースイベントが使用可能になると、それはすぐにによって処理されます。

source: ---1--2--3-4---5------- 
out: ------------4---5------- 

私は、動的更新のためのObservable.add()

var sourceStream = Observable.empty() 
var outStream = sourceStream.filter({x>3}) 

//print each element as its added 
sourceStream .subscribe({println(it)}) 
outStream.subscribe({println(it)}) 

for i in range(5): 
    sourceStream.add(i) 

を見つけることが期待される。このことは可能ですか？

Answer 1

私は新しいですが、どのように問題なしで問題を解決できますか？もし私が のアプリケーションをテストしていて、5秒ごとにアップデートを "ポップ"したいのであれば、この他の方法はどうすればいいですか？ ビジネスに登録してください？誰かが のサブスクライバを含まないこの質問への回答を投稿できますか？

5秒ごとに更新をポップアップし、interval演算子でObservableを作成する場合は、Subjectを使用しないでください。 Observablesを構築するために数十の異なる演算子がありますので、あなたはほとんど対象を必要としません。

と言われていますが、時には1つが必要です。テストコードの場合は、非常に便利です。私はユニットテストでそれらを広範囲に使用します。

To Use Subject Or Not To Use Subject?は被験者の主題に関する優れた記事です。