RxJava basic processing sequence (old)

Reactive study articles

I want to understand Reactive but I don't know anything I will describe how the sample code of Reactive Programming With RxJava works behind the scenes.

Since it's a 2016 book, it uses old APIs that are deprecated, such as Observable.create.

RxJava

Simple characters

simple

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.subscribe { s -> println("VALUE -> $s") }
}

• create {s -> s.OnNext("one") ...} --The argument lambda expression is interpreted as an implementation of OnSubscribe.
• subscribe { s -> println("VALUE -> $s") } --The argument lambda expression is an implementation of the internally used interface called Action1. --Wrap Action1 inside subscribe to create an instance of Subscriber --Then call call of OnSubscribe. --This calls the lambda expression of the argument of create. --The argument of call is the Subscriber created immediately before. --The lambda expression calls onNext for the argument Subscriber. --onNext of Subscriber is called by calling the lambda expression of the argument of subscribe.

doOnNext

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.doOnNext { i -> println("doOnNext $i") }
            .subscribe { s -> println("VALUE -> $s") }
}

The process in which orange is created by the doOnNext association You can see that it is acting like a proxy

map

fun main(args: Array<String>) {
    Observable.create<String> { s ->
        s.onNext("one")
        s.onCompleted()
    }.map { "hello $it" }
            .subscribe { s -> println("VALUE -> $s") }
}

Almost the same behavior as doOnNext

Publisher from another Thread

fun main(args: Array<String>) {
    println(Thread.currentThread())
    Observable.create<String> { s ->
        Thread {
            println(Thread.currentThread())
            s.onNext("one")
            s.onNext("two")
            s.onCompleted()
        }.start()
    }.subscribe { s -> println("VALUE -> $s by ${Thread.currentThread()}") }
}

//Output result
Thread[main,5,main]
Thread[Thread-0,5,main]
VALUE -> one by Thread[Thread-0,5,main]
VALUE -> two by Thread[Thread-0,5,main]
VALUE -> three by Thread[Thread-0,5,main]

If it is published in another thread, it will be subscribed in that thread.