withLatestFrom 运算

extension ObservableType {
    public func withLatestFrom<Second0: ObservableConvertibleType>(second: Second0)->Observable<Second0.E> {
        return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: { $ 1} )
    }

    public func withLatestFrom<Second0: ObservableConvertibleType, ResultType>(second: Second0, resultSelector: (E, Second0.E) throws->ResultType) ->Observable<ResultType> {
        return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: resultSelector)
    }
}

class WithLatestFrom<FirstType, SecondType, ResultType>: Producer<ResultType> {
    typealias ResultSelector = (FirstType, SecondType) throws -> ResultType

    private let _first: Observable<FirstType>
    private let _second: Observable<SecondType>
    private let _resultSelector: ResultSelector

    init(first: Observable<FirstType>, second: Observable<SecondType>, resultSelector: ResultSelector) {
        _first = first
        _second = second
        _resultSelector = resultSelector
    }

    override fun run<O: ObserverType where O.E = ResultType>(observer: O)->Disposable {
        let sink = WithLatestFromSink(parent: self, observer: observer)
        sink.disposable = sink.run()
        return sink
    }
}

class WithLatestFromSink<FirstType, SecondType, ResultType, O: ObserverType where O.E == ResultType>: Sink<O>, ObserverType, SynchronizedOnType, LockOwnerType {
    typealias Parent = WithLatestFrom<FirstType, SecondType, ResultType>
    typealias E = FirstType

    private let _parent : Parent

    var _lock = NSRecursiveLock()
    private var _latest: SecondType?

    init(parent: Parent, observer: O) {
        _parent = parent
        super.init(observer: observer)
    }

    func run()->Disposable {
        let sndSubscription = SingleAssignmentDisposable()
        let sndo = WithLatestFromSecond(parent: self, disposable: sndSubscription)

        sndSubscription.disposable = _parent._second.subscribe(sndo)
        let fstSubscription = _parent._first.subscribe(self)

        return StableCompositeDisposable.create(fstSubscription, sndSubscription)
    }

    func on(event: Event<E>) {
        synchronizedOn(event)
    }

    func _synchronized_on(event: Event<E>) {
        switch event {
        cae let .Next(value):
            guard let latest = _latest else {return}
            do {
                let res = try _parent._resultSelector(value, latest)
                forwardOn(.Next(res))
            }catch let e {
                forwardOn(.Error(e))
                dispose()
            }
        case .Completed:
            forwardOn(.Completed)
            dispose()
        case let .Error(error):
            forwardOn(.Error(error))
            dispose()
        }
    }
}    

class WithLatestFromSecond<FirstType, SecondType, ResultType, O: ObserverType where O.E == ResultType>: ObserverType, LockOwnerType, SynchronizedOnType {
    typealias Parent = WithLatestFromSink<FirstType, SecondType, ResultType, O>
    typealias E = SecondType

    private let _parent: parent
    private let _disposable: Disposable

    var _lock: NSRecursiveLock {
        get {
            return _parent._lock
        }
    }

    init(parent: parent, disposable: Disposable) {
          _parent = parent
          _disposable = disposable
    }

    func on(event: Event<E>) {
        _synchronized_on(event)
    }

    func _synchronized_on(event: Event<E>) {
        switch event {
        case let .Next(value):
            _parent._latest = value
        case .Completed:
            _disposable.dispose()
        case let .Error(error):
            _parent.forwardOn(.Error(error))
            _parent.dispose()
        }
    }
}