Stream to filter until another stream emits, but reemit those filtered emissions once 2nd stream emits

Question

I have a stream of Foo. Foo emits require Android View to be laid out (width and height > 0). I use RxBinding for that i.e.

fooOservable()
   .subscribe(foo -> {});

RxView.preDraws(mPager, () -> true)
                    .take(1);

When this observable emits (or completed, cuz of the take) view is laid out.

What I need is for fooObservable() to wait until RxView emits, i.e. view is laid out. I cant just fooObservable.filter( width && height > 0 ), because that drops the emits. What I need is to have the last emit of foo cached (if view not laid out) and reemitted after first RxView.preDraws., if it is laid out, it should pass normally

Answer 1

You can use delay or delaySubsription. There are forms which delay emitions or subscription until another observable emits something. In the first case you will have just emition delay, in the second case the whole subscription will be delayed.

Answer 2

I've written a sample using delay to wait another observable before emit your data.

class SimpleTest {
  val testScheduler = TestScheduler()

  @Test
  fun test() {
    fooObservable()
        .doOnNext { logger("Next", it.toString()) }
        .delay { Observable.timer(5, TimeUnit.SECONDS, testScheduler) }
        .subscribe { logger("Delayed", it.toString()) }

    testScheduler.advanceTimeBy(10, TimeUnit.SECONDS)
  }

  fun fooObservable(): Observable<Int> {
    return Observable.just(1, 2, 3, 4)
  }

  fun logger(tag: String, message: String): Unit {
    val formattedDate = Date(testScheduler.now()).format()
    System.out.println("$tag @ $formattedDate: $message")
  }

  fun Date.format(): String {
    return SimpleDateFormat("HH:mm:ss.SSS", Locale.US).format(this)
  }
}

I used a 5 seconds timer to simulate your secondary observable.

This code prints that output:

Next @ 21:00:00.000: 1
Next @ 21:00:00.000: 2
Next @ 21:00:00.000: 3
Next @ 21:00:00.000: 4
Delayed @ 21:00:05.000: 1
Delayed @ 21:00:05.000: 2
Delayed @ 21:00:05.000: 3
Delayed @ 21:00:05.000: 4