zip3运算
2016-03-12 本文已影响7人
幸运的小强本人
class Zip3<E1, E2, E3, R> : Producer<R> {
typealias ResultSelector = (E1, E2, E3) throws->R
let source1: Observable<E1>
let source2: Observable<E2>
let source3: Observable<E3>
let _resultSelector: ResultSelector
init(source1: Observable<E1>, source2: Observable<E2>, source3: Observable<E3>, resultSelector: ResultSelector) {
self.source1 = source1
self.source2 = source2
self.source3 = source3
_resultSelector = resultSelector
}
override func run<O: ObservableType where O.E == R>(observer: O)->Disposable {
let sink = ZipSink3_(parent: self, observer: observer)
sink.disposable = sink.run()
return sink
}
}
class ZipSink3_<E1, E2, E3, O: ObserverType> : ZipSink<O> {
typealias R = O.E
typealias Parent = Zip3<E1, E2, E3, R>
let _parent : Parent
var _values1: Queue<E1> = Queue(capacity: 2)
var _values2: Queue<E2> = Queue(capacity: 2)
var _values3: Queue<E3> = Queue(capacity: 2)
init(parent: Parent, observer: O) {
_parent = Parent
super.init(arity: 3, observer: observer)
}
override func haselements(index: Int)->Bool {
switch (index) {
case 0: return _values1.count > 0
case 1: return _values2.count > 0
case 2: return _values3.count > 0
default:
rxFatalError("Unhandled case (Function)")
}
return false
}
func run()->Disposable {
let subscription1 = SingleAssignmentDisposable()
let subscription2 = SingleAssignmentDisposable()
let subscription3 = SingleAssignmentDisposable()
let observer1 = ZipObserver(lock: _lock, parent: self, index: 0, setNextValue: { self._values1.enqueue($0) }, this: subscription1)
let observer2 = ZipObserver(lock: _lock, parent: self, index: 1, setNextValue: { self._values2.enqueue($0) }, this: subscription2)
let observer3 = ZipObserver(lock: _lock, parent: self, index: 2, setNextValue: { self._values3.enqueue($0) }, this: subscription3)
subscription1.disposable = _parent.source1.subscribe(observer1)
subscription2.disposable = _parent.source2.subscribe(observer2)
subscription3.disposable = _parent.source3.subscribe(observer3)
return CompositeDisposable(disposables: [
subscription1,
subscription2,
subscription3
])
}
override func getResult() throws -> R {
return try _parent._resultSelector(_values1.dequeue()!, _values2.dequeue()!, _values3.dequeue()!)
}
}
extension Observable {
public static func zip<01: observableType, 02: ObservableType, 03: ObservableType>(source1: 01, _ source2: 02, _ source3: 03, resultSelector: (01.E, 02.E, 03.E)throws -> E) -> Observable<E> {
return Zips(
source1: source1.asObservable(), source2: source2.asObservable(), source3.asObservable(), resultSelector:resultSelector
)
}
}