RxJava, RxKotlin - Observable (2) - Single , Maybe

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

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

(예제 code 들은 모두 Kotlin으로 작성하였습니다.)

 

Single

 

Observable은 무한한 데이터 발행이 가능하지만, Single 클래스는 오직 1개만 발행하도록 되어있습니다.

 

출처 : RxJava doc Single

 

이 클래스의 중요한 점은 발행과 동시에 종료된다는 점입니다. 종료시 onSuccess가 호출 됩니다.

 

just() 함수

Single 클래스는 Observable과 같은 방법으로 활용할 수 있습니다.

fun main(){
	val single = Single.just("hellow single")
    
    single.subscribe{data-> println(data)}
    //or
    single.subscribeBy(
        onSuccess = {println(it)}
    )
    //or
    single.subscribe(Consumer<String>(){t->println(t)})
}

등의 방법으로 Single을 만들 수 있습니다. 그리고 이런 Single은 Observable에서 활용 할 수 있습니다..

fun main(){

    //Single 클래스를 Observable로 부터 생성
    val source_1 = Observable.just("Hello Single first")
    Single.fromObservable(source_1).subscribe{it->println(it)}
    //Observable의 just에서 single 함수를 통한 생성
    Observable.just("Hello Single second").single("default item").subscribe{it->println(it)}

    //Observable의 first 함수를 통한 생성 첫번째 데이터이 후 onSuccess 호출 후 종료
    val colors = arrayListOf("Red","Blue","Green")
    colors.toObservable().first("default value").subscribe{it->println(it)}
    //빈 데이터에서 Single 호출 시 defaultItem에 지정한 데이터 호출
    Observable.empty<Any>().single("default Item").subscribe{it->println(it)}
    // take는 나중에 알아보겠지만 1번째 까지만 호출 하도록 합니다. 만약 take를 2로 하면 onError가 호출 됩니다. 
    Observable.just(Order("A_1"), Order("B_2")).take(1) .single(Order("Default Id")).subscribeBy(
        onError = {println("onError")},
        onSuccess = {println(it)}
    )
}

data class Order(val id : String){
    override fun toString(): String {
        return "Order id - $id"
    }
}

위의 take 에서도 이미 설명했지만 그래도 single 클래스에 2개의 데이터를 요청해보겠습니다.

 

fun main(){
    Observable.just("hello hello", "Error").single("default itme").subscribe{it->println(it)}
}

이렇게 할 경우 어떤 결과가 나올까요?

io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.lang.IllegalArgumentException: Sequence contains more than one element!
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
	at io.reactivex.internal.observers.ConsumerSingleObserver.onError(ConsumerSingleObserver.java:46)
	at io.reactivex.internal.operators.observable.ObservableSingleSingle$SingleElementObserver.onNext(ObservableSingleSingle.java:80)
	at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java:108)
	at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java:37)
	at io.reactivex.Observable.subscribe(Observable.java:12267)
	at io.reactivex.internal.operators.observable.ObservableSingleSingle.subscribeActual(ObservableSingleSingle.java:35)
	at io.reactivex.Single.subscribe(Single.java:3603)
	at io.reactivex.Single.subscribe(Single.java:3589)
	at io.reactivex.Single.subscribe(Single.java:3559)
	at Rx_single_testKt.main(rx_single_test.kt:32)
	at Rx_single_testKt.main(rx_single_test.kt)
Caused by: java.lang.IllegalArgumentException: Sequence contains more than one element!
	... 10 more
Exception in thread "main" io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.lang.IllegalArgumentException: Sequence contains more than one element!
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
	at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
	at io.reactivex.internal.observers.ConsumerSingleObserver.onError(ConsumerSingleObserver.java:46)
	at io.reactivex.internal.operators.observable.ObservableSingleSingle$SingleElementObserver.onNext(ObservableSingleSingle.java:80)
	at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java:108)
	at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java:37)
	at io.reactivex.Observable.subscribe(Observable.java:12267)
	at io.reactivex.internal.operators.observable.ObservableSingleSingle.subscribeActual(ObservableSingleSingle.java:35)
	at io.reactivex.Single.subscribe(Single.java:3603)
	at io.reactivex.Single.subscribe(Single.java:3589)
	at io.reactivex.Single.subscribe(Single.java:3559)
	at Rx_single_testKt.main(rx_single_test.kt:32)
	at Rx_single_testKt.main(rx_single_test.kt)
Caused by: java.lang.IllegalArgumentException: Sequence contains more than one element!
	... 10 more

위와 같은 error가 나옵니다.

이유는 위에 바로 보이는 것 처럼 Sequence contains more than one element라고 적혀있습니다. 즉 첫번째 onNext 후 두번째 onNext로 데이터 발행할 때 에러가 발생합니다.

 

Maybe

 

Maybe는 Single 클래스와 비슷합니다. 최대 데이터 하나를 가질 수 있기 때문이죠. 하지만 데이터 발행 없이 바로 데이터 발생을 완료 할 수 있습니다. 즉, Maybe 클래스는 Single 클래스에 onComplete 이벤트가 추가된 형태 입니다.

Maybe의 마블 다이어 그램

초록색 check 모양은 onSuccess 이벤트 x는 onError, ㅣ은 onComplete 이벤트에 해당합니다.

 

Maybe 객체는 Maybe 클래스로 생성할 수 있지만, 보통 Observable의 특정 연사자를 통해 생성할 떄가 많습니다.

또한 Maybe 객체를 생성할 수 있는 리액티브 연산자에는 elementAt(), firstElement() 등 여러가지가 있지만 이에 대해서는 추후 다뤄보도록 하겠습니다.