river.swift

import UIKit
import Async
import RxSwift

class Store {
    func get(river: River)(param: Double) -> Observable<Int64> {
        let s1 = interval(param, MainScheduler.sharedInstance)
        return merge(returnElements(s1, river.action)) >- debug("debug")
    }
}

class River {
    static let instance = River()

    let action = PublishSubject<Int64>()
    var getStore: ((param: Double) -> Observable<Int64>)!

    init() {
        getStore = Store().get(self)
    }
}

class ViewController: UIViewController {
    override func viewDidLoad() {

        let d = River.instance.getStore(param: 1)
            >- subscribeNext(println)

        sendNext(River.instance.action, 123)

        Async.main(after: 4) {
            d.dispose()
        }

    }
}