RxSwiftでHotなストリームにsubscribe(on:)してみる

gistfile1.swift

import RxSwift
import Foundation

let queue = DispatchQueue.global(qos: .background)

// 念の為に書くと、ここはメインスレッドで呼び出している
let stream = Observable.just(1)
    .flatMap { value -> Observable<Int> in
        print("flatMap: isMain \(Thread.isMainThread)")

        return .create { observer in
            // 通信ライブラリの結果をObservableにしたものというイメージ
            queue.async {
                observer.onNext(value * 10)
            }
            return Disposables.create()
        }
    }
    .share(replay: 1)

stream
    .map { value -> Int in
        print("map:, isMain  \(Thread.isMainThread)")
        return value * 100
    }
    .subscribe(on: ConcurrentDispatchQueueScheduler(queue: queue))
    .observe(on: MainScheduler.instance)
    .subscribe(onNext: {
        print("onNext: \($0) isMain \(Thread.isMainThread)")
    })

stream
    .map { value -> Int in
        print("map2:, isMain  \(Thread.isMainThread)")
        return value * 100
    }
    .subscribe(on: MainScheduler.instance)
    .observe(on: MainScheduler.instance)
    .subscribe(onNext: {
        print("onNext2: \($0) isMain \(Thread.isMainThread)")
    })

• Hotだからコントロールできないわけじゃなく、flatMapは別スレッドで動いている
• subscribe順にコントロールしてるように見る
• 2つ目のsubscribeでは意味がないことがわかるし、mapまでも別スレッドで動いてしまっている