[RxJava] Observable 클래스

 

rajava 프로그래밍 책을 읽으면서 공부중입니다. 시간이 부족해 매번 같은 곳만 보고 있네요... 간략히 정리좀 해볼려고합니다.

 

제목은 말 그대로 옵서버 패턴을 구현한 클래스입니다. 옵서버 패턴은 객체의 상태 변화를 관찰하는 옵서버 목록을 객체에 등록합니다.

Observable 클래스를 통해 데이터 발행 이벤트를 추가하고 subscribe() 함수를 통해 Observable을 구독하면 그 때부터 구독자에게 데이터가 발행되기 시작합니다. 

옵서버 패턴은 유튜버와 구독자의 개념으로 보면 이해하기 쉽습니다. 구독자가 유튜버 채널을 구독하면 유튜버가 영상을 올릴 때마다 구독자에게 알림이 가서 알 수 있습니다.

 

[알림 이벤트]

Observable은 3가지 알림 메소드가 있습니다.

onNext : Observable 데이터의 발행을 알립니다.

onComplete: 모든 데이터가 발행됬음을 알립니다. 한번만 호출되며 모든 데이터가 발행됬으므로 onNext 이벤트가 발생하지 않습니다.

onError : Observable에 에러가 발생했음을 알립니다. 이 이후에는 onNext, onComplete 이벤트가 발생하지 않고 Observable 실행이 종료됩니다.

 

 

[발행 이벤트]

just() : 넣는 순서대로 그냥 발행합니다. 가장 기본적인 구조입니다.

Observable.just(1,2,3,4)
	.subscribe(System.out::println)
        
 1
 2
 3
 4

 

 

create() : 함수에 데이터를 넣으면 자동으로 알림 이벤트가 발생하는 just() 와 달리 onNext, onComplete, onError 같은 알림을 개발자가 직접 호출합니다.

Observable.create<String> { emitter ->
        emitter.onNext("1")
        emitter.onNext("2")
        emitter.onNext("3")
        emitter.onComplete()
}.subscribe({
        Log.d("onNext", it.toString())
 	}, {
        Log.d("onError", it.message)
	}, {
        Log.d("onComplete", "Complete")
    })
    
 Observable<String> source = Observable.create(
 	(ObservableEmitter<String> emitter) -> {
        emitter.onNext("1")
        emitter.onNext("2")
        emitter.onNext("3")
        emitter.onComplete()
});

source.subscribe(data -> Lod.d("data", data))
 

 

 

 

fromArray() : just(), create() 와 달리 배열 데이터를 처리할 때 사용합니다. 

 

fromIterable() : Iterable 인터페이스를 구현한 클래스 데이터를 처리할 때 사용합니다. 대표적으로 ArrayList, HashSet, LinkedList, TreeSet, Vector 등이 있습니다.

 

fromCallable() : 비동기 작업에 사용됩니다. 자바 5에서 추가된 동시성 API인 Callable 인터페이스 객체와 함께 사용됩니다. (비동기 클래스)

Callable<String> callable = ()->{
            Thread.sleep(1000);
            return "Hello Callable";
 };
 Observable<String> source=Observable.fromCallable(callable);
 source.subscribe(System.out::println);
 
 //출력 Hello Callable

 

fromFuture() : 마찬가지로 자바 5에서 추가된 동시성 API 중 하나인 Future 인터페이스 객체와 함께 사용한다는데 실제 사용법은 아직 잘 모르겠습니다. 뒤에서 스케줄러와 함께 활용되는 경우가 많다는데 나중에 보도록 합니다. fromCallable() 과 다른점은 반환된 Future 객체에 get() 메서드를 호출하면 결괏값이 나올 때까지 기다린다는 점만 짚고 넘어갑니다.

 

[구독관련]

subscirbe() : Rxjava는 내가 동작시키기 원하는 것을 사전에 정의 후 실제 그것이 실행되는 시점을 subscribe() 함수를 통해 조절할 수 있습니다. 위으 just() 함수의 경우에도 팩토리 함수로 구현되어 데이터 흐름을 정의 한 후 subscribe() 함수를 호출해야 데이터가 발행됩니다. 

 

Disposable : subscribe() 는 Disposable 이라는 인터페이스 객체를 반환합니다. 이것은 구독 객체에 해당되며 다음과 같은 함수를 갖고 있습니다.

void dispose()
=> Observable에게 더이상 데이터를 발행하지 않도록 구독을 해지합니다.
=> Observable이 onComplete 알림을 보내면 자동으로 이 함수가 호출되 구독을 해지해줍니다.
boolean isDisposed()
=> 말그대로 구독이 해지되었는지(데이터 발행하지 않는지) 확인하는 함수입니다.

 

 

 

 

 

 

[커니의 코틀린 RxJava Retrofit2 예제코드]

package com.androidhuman.example.simplegithub.ui.repo

import android.os.Bundle
import android.support.v7.app.AppCompatActivity
import android.view.View
import com.androidhuman.example.simplegithub.R
import com.androidhuman.example.simplegithub.api.provideGithubApi
import com.androidhuman.example.simplegithub.extensions.plusAssign
import com.androidhuman.example.simplegithub.ui.GlideApp
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.CompositeDisposable
import kotlinx.android.synthetic.main.activity_repository.*
import java.text.ParseException
import java.text.SimpleDateFormat
import java.util.Locale

class RepositoryActivity : AppCompatActivity() {

    companion object {

        const val KEY_USER_LOGIN = "user_login"

        const val KEY_REPO_NAME = "repo_name"
    }

    internal val api by lazy { provideGithubApi(this) }

    internal val disposables = CompositeDisposable()

    internal val dateFormatInResponse = SimpleDateFormat(
            "yyyy-MM-dd'T'HH:mm:ssX", Locale.getDefault())

    internal val dateFormatToShow = SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss", Locale.getDefault())

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_repository)

        val login = intent.getStringExtra(KEY_USER_LOGIN) ?: throw IllegalArgumentException(
                "No login info exists in extras")
        val repo = intent.getStringExtra(KEY_REPO_NAME) ?: throw IllegalArgumentException(
                "No repo info exists in extras")

        showRepositoryInfo(login, repo)
    }

    override fun onStop() {
        super.onStop()
        disposables.clear()
    }

    private fun showRepositoryInfo(login: String, repoName: String) {
        disposables += api.getRepository(login, repoName)
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe { showProgress() }
                .doOnError { hideProgress(false) }
                .doOnComplete { hideProgress(true) }
                .subscribe({ repo ->
                    GlideApp.with(this@RepositoryActivity)
                            .load(repo.owner.avatarUrl)
                            .into(ivActivityRepositoryProfile)

                    tvActivityRepositoryName.text = repo.fullName
                    tvActivityRepositoryStars.text = resources
                            .getQuantityString(R.plurals.star, repo.stars, repo.stars)
                    if (null == repo.description) {
                        tvActivityRepositoryDescription.setText(R.string.no_description_provided)
                    } else {
                        tvActivityRepositoryDescription.text = repo.description
                    }
                    if (null == repo.language) {
                        tvActivityRepositoryLanguage.setText(R.string.no_language_specified)
                    } else {
                        tvActivityRepositoryLanguage.text = repo.language
                    }

                    try {
                        val lastUpdate = dateFormatInResponse.parse(repo.updatedAt)
                        tvActivityRepositoryLastUpdate.text = dateFormatToShow.format(lastUpdate)
                    } catch (e: ParseException) {
                        tvActivityRepositoryLastUpdate.text = getString(R.string.unknown)
                    }
                }) {
                    showError(it.message)
                }
    }

    private fun showProgress() {
        llActivityRepositoryContent.visibility = View.GONE
        pbActivityRepository.visibility = View.VISIBLE
    }

    private fun hideProgress(isSucceed: Boolean) {
        llActivityRepositoryContent.visibility = if (isSucceed) View.VISIBLE else View.GONE
        pbActivityRepository.visibility = View.GONE
    }

    private fun showError(message: String?) {
        with(tvActivityRepositoryMessage) {
            text = message ?: "Unexpected error."
            visibility = View.VISIBLE
        }
    }
}