[RxJava] Observable

1. just

import io.reactivex.Observable

class RxJavaExample {
    fun emit() {
        Observable.just("Hello", "RxJava 2!!", 123)
            .subscribe { data ->
                println(data)
            }
    }
}

fun main() {
    val rxJavaExample = RxJavaExample()
    rxJavaExample.emit()
}

실행 결과

 

 

 

 

import io.reactivex.Observable

fun main() {
    val observable = Observable.just("A", "B", "C")
    val disposable = observable.subscribe(
        { v -> println(v) },
        { err -> println(err) },
    ) { println("complete") }

    println(disposable.isDisposed)
}

실행 결과

 

 

 

 

2. create

import io.reactivex.Observable

fun main() {
    val observable = Observable.create<String> {
        it.onNext("A")
        it.onNext("B")
        it.onNext("C")
        it.onComplete()
    }

    observable.subscribe {
        println(it)
    }
}

실행 결과

 

 

 

 

 

3. fromArray

import io.reactivex.Observable

fun main() {
    val arr = arrayOf("A", "B", "C")
    val observable = Observable.fromArray(arr)

    observable.subscribe {
        it.map { element ->
            println(element)
        }
    }
}

실행 결과

 

 

 

4. fromIterable

import io.reactivex.Observable

fun main() {
    val arrList = arrayListOf<String>()
    arrList.add("A")
    arrList.add("B")
    arrList.add("C")
    val observable = Observable.fromIterable(arrList)

    observable.subscribe {
        it.map { element ->
            println(element)
        }
    }
}

실행 결과

 

 

 

import io.reactivex.Observable

fun main() {
    val hashSet = hashSetOf<String>()
    hashSet.add("C")
    hashSet.add("A")
    hashSet.add("B")
    val observable = Observable.fromIterable(hashSet)

    observable.subscribe {
        it.map { element ->
            println(element)    // 정렬돼서 출력
        }
    }

    val mutableSet = mutableSetOf<String>()
    mutableSet.add("C")
    mutableSet.add("A")
    mutableSet.add("B")
    val observable2 = Observable.fromIterable(mutableSet)

    observable2.subscribe {
        it.map { element ->
            println(element)    // 넣은 순서대로 출력
        }
    }

}

실행 결과

 

 

 

import io.reactivex.Observable
import java.util.concurrent.ArrayBlockingQueue
import java.util.concurrent.BlockingQueue

fun main() {

    val queue: BlockingQueue<String> = ArrayBlockingQueue(3)
    queue.put("B")
    queue.put("C")
    queue.put("A")

    val observable = Observable.fromIterable(queue)
    observable.subscribe {
        println(it)
    }

}

실행 결과

 

 

 

 

5. fromCallable