RxJava, RxKotlin - skipUntil 함수

[출처 - RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에]

본 글은 'RxJava 프로그래밍 : 리액티브 프로그래밍 기초부터 안드로이드 까지 한번에' 를 학습하면서 정리한 글입니다

 

마블다이어그램 참고 -http://reactivex.io/RxJava/javadoc/io/reactivex/Flowable.html#skipUntil-org.reactivestreams.Publisher-

 

skipUntil 함수는 데이터 발행을 다른 Observable에서 특정 값이 들어오는 그 순간 부터 데이터를 발행합니다.

그럼 test code를 확인해 보겠습니다.(  CommonUtils와 CommonUtilsk의 내용은 여기에서 확인 할 수 있습니다.)
Java

import io.reactivex.Observable;

import java.util.concurrent.TimeUnit;

public class skipUntil_test {

    public static void main(String[] args){

        String[] data = {"1","2","3","4","5","6"};
        CommonUtils.exampleStart();
        Observable source = Observable.fromArray(data).zipWith(Observable.interval(100L, TimeUnit.MILLISECONDS), (val, notUsed)-> val)
                .skipUntil(Observable.timer(500L,TimeUnit.MILLISECONDS)).doOnComplete(()->System.out.println("onComplete"));

        source.subscribe(System.out::println);

        CommonUtils.sleep(1000);

    }
}

 

Kotlin

import io.reactivex.Observable
import java.util.concurrent.TimeUnit
import io.reactivex.functions.BiFunction
import io.reactivex.rxkotlin.toObservable

fun main(){

    val data = arrayOf("1", "2", "3", "4", "5", "6")
    CommonUtils.exampleStart()
    val source = data.toObservable()
        .zipWith<Long, String>(Observable.interval(100L, TimeUnit.MILLISECONDS), BiFunction { t1, t2 -> t1 })
        .skipUntil(Observable.timer(500L, TimeUnit.MILLISECONDS)).doOnComplete({ println("onComplete") })

    source.subscribe{it->println(it)}

    CommonUtils.sleep(1000)

}

결과

5
6
onComplete

 

12,3,4,5,6을 100ms마다 발행을 하는데 500ms의 timer를 주었기 때문에 결과에 5,6인 나타나고 모든 결과를 발행했기 때문에 onComplete가 나타납니다.