在 Rx 里面，實現(xiàn)的想法是 有一個數(shù)據(jù)源頭鳄橘，然后觀察者去關(guān)注，得到數(shù)據(jù)的回調(diào)术徊，中間可能通過很多鏈式的操作符鲸湃，簡單優(yōu)美。
Observable 是 數(shù)據(jù)源笋除，Subscriber Observer 是觀察者垃它。
Observable 構(gòu)造方法有個參數(shù) OnSubscribe，這個接口需要實現(xiàn)一個 call 方法洛史，在里面執(zhí)行數(shù)據(jù)庫查詢贝奇、 網(wǎng)絡(luò)請求之類的行為，并在里面 調(diào)用 Subscriber 的 onNext onCompleted onError。
基于上面的知識 我們可以完成一個初級版的demo陕习。

• 新建觀察者 數(shù)據(jù)源

//Observer.kt
interface Observer<T> {
    fun onCompleted()
    fun onError(t : Throwable)
    fun onNext(t: T)
}

//Subscription.kt
interface Subscription {
    fun unsubscribe()
    fun isUnsubscribed(): Boolean
}

//Subscriber.kt
abstract class Subscriber<T> : Observer<T>, Subscription{

    private var subscribe = true

    open fun onStart() {}

    override fun unsubscribe() {
        this.subscribe = false
    }

    override fun isUnsubscribed(): Boolean  = !subscribe
}

//Observable.kt
class Observable<T>(private var onSubscribe: OnSubscribe<T>?) {
    interface OnSubscribe<T> {
        fun call(subscriber: Subscriber<T>)
    }
}

• create 该镣， just响谓， subscribe

//Observable.kt
class Observable<T>(private var onSubscribe: OnSubscribe<T>?) {

    companion object {

        fun <T> create(subscribe: OnSubscribe<T>): Observable<T> {
            return Observable(subscribe)
        }

        fun <T> just(list: List<T>): Observable<T> {
            return create(object : OnSubscribe<T> {
                override fun call(subscriber: Subscriber<T>) {
                    try {
                        for (t in list) {
                            if (!subscriber.isUnsubscribed())
                                subscriber.onNext(t)
                            else
                                break
                        }
                    }catch (e: Exception) {
                        subscriber.onError(e)
                    }
                    subscriber.onCompleted()
                }
            })
        }

        fun <T> just(vararg ts : T): Observable<T> {
            return just(ts.asList())
        }
    }

    fun subscribe(observer: Observer<T>): Subscription{
        return if (observer is Subscriber)
            subscribe(observer)
        else {
            val subscriber = ObserverSubscriber(observer)
            subscribe(subscriber)
        }
    }

    fun subscribe(subscriber: Subscriber<T>): Subscription{
        subscriber.onStart()
        onSubscribe?.call(subscriber)
        return subscriber
    }
}

//ObserverSubscriber.kt
class ObserverSubscriber<T>(private var observer: Observer<T>) : Subscriber<T>(){

    override fun onCompleted() {
        observer.onCompleted()
    }

    override fun onError(t: Throwable) {
        observer.onError(t)
    }

    override fun onNext(t: T) {
        observer.onNext(t)
    }
}

 //UnitTest.kt
@Test
fun rx() {
    Observable.just(1, 2, 3)
        .subscribe(object : Subscriber<Int>() {
            override fun onCompleted() {

            }

            override fun onError(t: Throwable) {
            }

            override fun onNext(t: Int) {
                System.out.println(t)
            }
        })
}

map嫁审， filter

//OnSubscribeFilter.kt
class OnSubscribeFilter<T>(private var source: Observable<T>, private var predicate: (T) -> Boolean) : Observable.OnSubscribe<T> {

    override fun call(subscriber: Subscriber<T>) {
        source.subscribe(object : Subscriber<T>(){
            override fun onCompleted() {
                subscriber.onCompleted()
            }

            override fun onError(t: Throwable) {
                subscriber.onError(t)
            }

            override fun onNext(t: T) {
                val result = predicate(t)
                if (result)
                    subscriber.onNext(t)
            }
        })
    }
}

//OnSubscribeMap.kt
class OnSubscribeMap<T, R>(private var source: Observable<T>, private var transformer: (T) -> R) : Observable.OnSubscribe<R> {

    override fun call(subscriber: Subscriber<R>) {
        source.subscribe(object : Subscriber<T>(){
            override fun onCompleted() {
                subscriber.onCompleted()
            }

            override fun onError(t: Throwable) {
                subscriber.onError(t)
            }

            override fun onNext(t: T) {
                subscriber.onNext(transformer(t))
            }
        })
    }
}

//Observable.kt
fun <R> map(transformer: (t: T) -> R): Observable<R> {
    return create(OnSubscribeMap(this, transformer))
}
//Observable.kt
fun filter(predicate: (t: T) -> Boolean): Observable<T> {
    return create(OnSubscribeFilter(this, predicate))
}

//UnitTest.kt
@Test
fun rx() {
    Observable.just("1", "2", "3")
        .map {
            it.toInt() + 1
        }
        .filter {
            it != 1
        }
        .subscribe(object : Subscriber<Int>() {
            override fun onCompleted() {

            }

            override fun onError(t: Throwable) {
            }

            override fun onNext(t: Int) {
                System.out.println(t)
            }
        })

至上 完成了最基本的 Rx 的 操作符遏插，下篇將說到 線程切換方面的操作符。