[RxJava] Observable 생성하기

0. Observable,  그래서 어떻게 쓰이는데?

Observable은 다양한 종류의 이벤트나 데이터를 비동기적으로 방출할 수 있는 객체이다. Observable은 데이터 스트림을 나타내며, 이 Observable이 방출하는 이벤트나 데이터는 다양할 수 있는데 어떤 경우에 쓰이는지 살펴보자. 

 

• 버튼 클릭 이벤트: 사용자가 화면의 버튼을 클릭할 때 마다 Observable은 클릭 이벤트를 방출할 수 있다.
• 네트워크 요청 결과: 네트워크 요청을 통해 데이터를 받아올 때, Observable은 이 데이터를 방출할 수 있다. 비동기적인 HTTP 요청과 같은 작업에 유용하다.
• 타이머 또는 주기적인 작업: 일정 시간 간격으로 이벤트를 방출하는 Observable을 생성할 수 있다. 예를 들어, 1초마다 현재 시간을 방출하는 Observable을 만들 수 있다.
• 데이터베이스 변경 감지: 데이터 베이스의 특정 테이블에서 변경 사항이 발생할 때 Observable은 이 변경 사항을 감지하고 방출할 수 있다. 
• 사용자 입력: 사용자의 키보드 입력 또는 화면 터치 이벤트와 같은 사용자 입력을 Observable로 변환하여 처리할 수 있다.

이렇듯 Observable은 이러한 이벤트나 데이터의 스트림을 나타낸다. 

Observable을 생성하는 다양한 방법 중 기본적이면서 가장 많이 쓰는 것들을 살펴보면서 어떻게 생성하는지를 본다.

 

 

1. Observable 만들기

create()

You should call the ObservableEmitter's onNext, onError and onComplete methods in a serialized fashion. The rest of its methods are thread-safe.

 

Observable.create() 메소드를 쓰기 위해서는 개발자가 Emitter를 이용하여 직접 데이터의 발행을 해야한다. 예를 들어 데이터를 발행하려면 onNext() 메서드를 직접 호출하여 구현해야하며 모든 데이터를 발행한 후에는 반드시 onComplete() 메소드를 호출해야한다.

Observer는 subscribe 메소드를 이용하여 Observable을 구독하고 데이터를 전달받는다. 

• emitter : Observable의 구독자(Subscriber)에게 데이터나 이벤트를 전달하는 역할을 하는 객체. 알람을 주는 초인종과 같은 역할!
  

init {

    val observable = Observable.create<Int> { emitter ->
        try {
            // 데이터 방출
            emitter.onNext(1)
            emitter.onNext(2)
            emitter.onNext(3)

            // 작업 완료를 나타내는 onComplete 이벤트 방출
            emitter.onComplete()
        } catch (e: Exception) {
            // 에러 처리
            emitter.onError(e)
        }
    }

    // 구독자(subscriber) 등록 및 데이터 처리
    observable.subscribe(
        { value -> println("onNext: $value") },
        { error -> println("onError: $error") },
        { println("onCompleted") }
    )
   // 
    1
    2
    3
    onCompleted
    
}

 

just()

데이터를 있는 그대로 발행하는 Observable 을 생성한다. create() 과 달리 발행을 위한 Emitter를 직접 쓸 필요가 없다.

 

init {

    val observable: Observable<String> = Observable.just("1", "2", "3")

    observable.subscribe(
        { value -> println("onNext: $value") },
        { error -> println("onError: $error") },
        { println("onCompleted") }
    )
    //
    1
    2
    3
    onCompleted

}

 

from

convert various other objects and data types into Observables

 

주어진 데이터 소스가 있다면 from 으로 시작하는 연산자를 통해 Observable을 생성하고 데이터를 방출할 수 있다.

 

fromIterable()

주어진 소스가 컬렉션 또는 배열일 경우에 유용하게 쓰일 수 있다. 나는 예를 들어 많은 양의 데이터를 서버로부터 받아와야할 때 Iterable 인터페이스를 기반으로 하나씩 순회하면서 요청하고 응답을 받을 경우에 썼다.  컬렉션 또는 배열을 순서대로 Observable로 바꾸어 생성한다. 

 

init {

    val items = listOf("1", "2", "3")
    val observable = Observable.fromIterable(items)
    // 리스트 내 문자열을 순서대로 방출하는 Observable을 생성
    1
    2
    3
    
}

 

fromCallable()

fromCallable은 Callable을 사용하여 Observable을 생성하고 이 Observable을 구독하면 Callable에서 반환한 값을 비동기적으로 받을 수 있다. Exception을 발생시킬 수도 있다. 구독자 콜백을 통해 이러한 상황을 처리할 수 있다. 

 

val observable = Observable.fromCallable {
    // 비동기 작업 수행 후 결과를 반환
    fetchDataFromNetwork()
}
// fetchDataFromNetwork()를 실행하고 결과 값을 방출하는 Observable을 생성

 

range()

간단한 숫자 시퀀스 생성

just()와 같이 방출할 범위의 값을 입력하면 순서대로 방출한다. 

init {
    val observable = Observable.range(1, 5)
    observable.subscribe(
        { println(it)},
        { println("onError") },
        { println("onCompleted") }
    )
    // 1부터 5까지의 정수를 순서대로 방출하는 Observable을 생성
    1
    2
    3
    4
    5
    onCompleted
}

 

Interval

주기적인 이벤트 생성

val observable = Observable.interval(1, TimeUnit.SECONDS)
// 1초 간격으로 Long 값(시간 경과)을 방출하는 Observable을 생성

 

사용자 입력 또는 UI 이벤트 처리

val button = findViewById<Button>(R.id.button)
val observable = RxView.clicks(button)
// 버튼 클릭 이벤트를 감지하여 View 객체를 방출하는 Observable을 생성

 

Subject

Subject를 사용한 사용자 정의 Observable 생성

Subject는 데이트 스트림을 생성하고(Observable) 이를 구독할 수 있는 Observer의 역할을 모두 수행할 수 있는 특별한 형태의 객체인데, 이 Subject에 대해서는 다음편에서 다루도록 하겠다.

val subject = PublishSubject.create<String>()
// 사용자 정의로 이벤트를 방출할 수 있는 PublishSubject 생성
subject.onNext("A")
subject.onNext("B")
subject.onComplete()
// "A", "B"를 방출하고 작업을 완료하는 Observable을 생성