Rxは数秒が経過するまでスキップします

Question

私はX秒ごとに信号を放出し続けるPublishSubjectを持っていますが、Y秒後に放出される最初の項目についてのみ検討したいと思います。

例

• 観察Aは毎秒
• 観察Bのみ-間に ものを無視して、この "ダニ" 5秒ごとに発するべき "カチカチ" 発します。

つまり、一定の時間が経過するまで、すべての項目をスキップします。

これが原因デバウンスはは発光せずに5秒滞在する観察可能なオリジナルが必要ですが、鉱山は毎秒放出され、デバウンスないです。

Answer 1

私の解決策を見てください。 'tickEverySecond'は毎秒値を出力します。 '結果'は、5秒の時間枠の間に 'tickEverySecond'によって放出されるすべての項目を収集します。 5秒ごとに 'tickEverySecond'から最後に出力された値がサブスクライバにプッシュされます。

@Test 
    public void name() { 
    TestScheduler testScheduler = new TestScheduler(); 
    Flowable<Long> tickEverySecond = Flowable.interval(0, 1, TimeUnit.SECONDS, testScheduler); 

    Flowable<Long> result = tickEverySecond.window(5, TimeUnit.SECONDS, testScheduler).flatMap(longFlowable -> longFlowable.takeLast(1)); 

    TestSubscriber<Long> test = result.test(); 

    testScheduler.advanceTimeTo(4, TimeUnit.SECONDS); 
    test.assertValueCount(0).assertNotComplete(); 

    testScheduler.advanceTimeTo(5, TimeUnit.SECONDS); 
    test.assertValue(4L).assertNotComplete(); 

    testScheduler.advanceTimeTo(9, TimeUnit.SECONDS); 
    test.assertValues(4L).assertNotComplete(); 

    testScheduler.advanceTimeTo(10, TimeUnit.SECONDS); 
    test.assertValues(4L, 9L).assertNotComplete(); 

    testScheduler.advanceTimeTo(14_999, TimeUnit.MILLISECONDS); 
    test.assertValues(4L, 9L).assertNotComplete(); 
    } 

あなたは、次の演算子と同じ効果を持つことができます。

Flowable<Long> result = tickEverySecond.throttleLast(5, TimeUnit.SECONDS, testScheduler); 

Answer 2

は、私が思う.window（オペレータは何が必要です） - https://github.com/ReactiveX/RxSwift/blob/master/RxSwift/Observables/Window.swift

それはタイムフレームでイベントを分割し、 RxJavaのドキュメントを参照してください。 - http://reactivex.io/RxJava/javadoc/rx/Observable.html#window-long-long-java.util.concurrent.TimeUnit-