Observables.combineLatest vs Singles.zip (rxkotlin)

gistfile1.txt

    @Test
    fun withCombineLatest() {
        val testScheduler = TestScheduler()

        val observable1 = Single.defer {
            Single.just(1)
                .delay(1, TimeUnit.SECONDS, testScheduler)
        }.toObservable()
        val observable2 = Single.defer {
            Single.just(2)
                .delay(2, TimeUnit.SECONDS, testScheduler)
        }.toObservable()
        val observable3 = Single.defer {
            Single.just(3)
                .delay(3, TimeUnit.SECONDS, testScheduler)
        }.toObservable()

        val testObserver = Observables.combineLatest(observable1, observable2, observable3)
            .subscribeOn(testScheduler)
            .observeOn(testScheduler)
            .test()

        testObserver.assertNotTerminated()
            .assertNoErrors()
            .assertValueCount(0);

        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertNoValues()
        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertNoValues()
        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertValueCount(1)
    }

    @Test
    fun withZip() {
        val testScheduler = TestScheduler()

        val single1 = Single.defer {
            Single.just(1)
                .delay(1, TimeUnit.SECONDS, testScheduler)
        }
        val single2 = Single.defer {
            Single.just(2)
                .delay(2, TimeUnit.SECONDS, testScheduler)
        }
        val single3 = Single.defer {
            Single.just(3)
                .delay(3, TimeUnit.SECONDS, testScheduler)
        }

        val testObserver = Singles.zip(single1, single2, single3)
            .subscribeOn(testScheduler)
            .observeOn(testScheduler)
            .test()

        testObserver.assertNotTerminated()
            .assertNoErrors()
            .assertValueCount(0);

        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertNoValues()
        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertNoValues()
        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertValueCount(1)
    }

    @Test
    fun withCombineLatestTrueObservables() {
        val testScheduler = TestScheduler()

        val observable1 = Observable.defer {
            Observable.intervalRange(1, 2, 1L, 1L, TimeUnit.SECONDS, testScheduler)
        }

        val observable2 = Observable.defer {
            Observable.intervalRange(1, 2, 2L, 1L, TimeUnit.SECONDS, testScheduler)
        }

        val observable3 = Observable.defer {
            Observable.intervalRange(1, 2, 3L, 1L, TimeUnit.SECONDS, testScheduler)
        }

        val testObserver = Observables.combineLatest(observable1, observable2, observable3)
            .subscribeOn(testScheduler)
            .observeOn(testScheduler)
            .test()

        testObserver.assertNotTerminated()
            .assertNoErrors()
            .assertValueCount(0);

        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors().assertNoValues() //user gets nothing because only 1st observable emitted

        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors()
            .assertNoValues() //user gets nothing because only 1st emitted and 2nd observable starts to emit

        testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS);
        testObserver.assertNoErrors()
            .assertValueCount(2) //3rd starts to emit, user would get (2,1,1) and then (2,2,1)

        testScheduler.advanceTimeBy(
            1L,
            TimeUnit.SECONDS
        ) //last value of 3rd observable sequence arrives

        testObserver.assertNoErrors().assertValueCount(3) //user would get (2,2,2)
        testObserver.assertNoErrors().assertComplete()
    }