zip2运算

class Zip2<E1, E2, R> : Producer<R> {
typealias ResultSelector = (E1, E2) throws ->R

    let source1 = Observable<E1>
    let source2 = Observable<E2>

    let _resultSelector: ResultSelector

    init(source1: Observable<E1>, source2: Observable<E2>, resultSelector: ResultSelector) {
        self.source1 = source1
        self.source2 = source2

        _resultSelector = resultSelector
    }

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

class ZipSink2_<E1, E2, O: ObserverType> : ZipSink<O> {
    typealias R = O.E
    typealias Parent = Zip2<E1, E2, R>

    let _parent: Parent

    var _values1: Queue<E1> = Queue(capacity: 2)
    var _values2: Queue<E2> = Queue(capacity: 2)

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

    override func hasElements(index: Int)->Bool {
        switch(index) {
        case 0: return _values1.count > 0
        case 1: return _values2.count > 0

        default:
            rxFatalError("Unhandled case (Function)")
        }

        return false
    }

    func run()->Disposable {
        let subscription1 = SingleAssignmentDisposable()
        let subscription2 = SingleAssignmentDisposable()

        let observer1 = ZipObserver(lock: _lock, parent: self, index: 0, setNextValue: { self._values1.enqueue($0) }, this: subscription)
        let observer2 = ZipObserver(lock: _lock, parent: self, index: 1, setNextValue: {self._values2.enqueue($0) }, this: subscription2)

        subscription1.disposable = _parent.source1.subscribe(observer)
        subscription2.disposable = _parent.subscribe(observer)

        return CompositeDisposable(disposables: [
            subscription1,
            subscription2
        ])
    }

    override func getResult() throws->R {
        return try _parent._resultSelector(_values1.dequeue()!, _values2.dequeue()!)
    }
}

extension Observable {
    /**
    Merges the specified observable sequences into 
    one observable sequence by using the selector 
    function whenever all of the observable sequence 
    have producted an element at a corresponding index.

    - parameter resultSelector: Function to invoke for each series of elements at corresponding indexes in the sources.
    - returns: An observable sequence containing the result of combining elements of the sources using the specified result selector function.
    */
    @warn_unused_result(message="http://git.io/rxs.uo")
    public static funczip<01: ObservableType, 02: ObservableType>(source1: 01, _ source2: 02, resultSelector: (01.E, 02.E) throws ->E)->Observable<E> {
        return Zip2(
            source1: source1.asObservable(), source2: source2.asObservable(), resultSelector: resultSelector
        )
    }
}