0001    //
0002    //  Zip+CollectionType.swift
0003    //  Rx
0004    //
0005    //  Created by Krunoslav Zaher on 8/30/15.
0006    //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
0007    //
0008    
0009    import Foundation
0010    
0011    class ZipCollectionTypeSink
Zip+CollectionType.swift:133
        let sink = ZipCollectionTypeSink(parent: self, observer: observer)
<C
Zip+CollectionType.swift:13
    typealias Parent = ZipCollectionType<C, R>
Zip+CollectionType.swift:14
    typealias SourceElement = C.Generator.Element.E
: CollectionType, R
Zip+CollectionType.swift:13
    typealias Parent = ZipCollectionType<C, R>
, O
Zip+CollectionType.swift:12
    : Sink<O> {
Zip+CollectionType.swift:27
    init(parent: Parent, observer: O) {
: ObserverType where C.Generator.Element : ObservableConvertibleType, O.E == R>
0012        : Sink<O> {
0013        typealias Parent
Zip+CollectionType.swift:16
    private let _parent: Parent
Zip+CollectionType.swift:27
    init(parent: Parent, observer: O) {
 = ZipCollectionType<C, R>
0014        typealias SourceElement
Zip+CollectionType.swift:22
    private var _values: [Queue<SourceElement>]
Zip+CollectionType.swift:29
        _values = [Queue<SourceElement>](count: parent.count, repeatedValue: Queue(capacity: 4))
Zip+CollectionType.swift:41
    func on(event: Event<SourceElement>, atIndex: Int) {
Zip+CollectionType.swift:61
                    var arguments = [SourceElement]()
 = C.Generator.Element.E
0015        
0016        private let _parent
Zip+CollectionType.swift:28
        _parent = parent
Zip+CollectionType.swift:51
                if _numberOfValues < _parent.count {
Zip+CollectionType.swift:53
                    if numberOfOthersThatAreDone == _parent.count - 1 {
Zip+CollectionType.swift:62
                    arguments.reserveCapacity(_parent.count)
Zip+CollectionType.swift:74
                    let result = try _parent.resultSelector(arguments)
Zip+CollectionType.swift:93
                if _numberOfDone == _parent.count {
Zip+CollectionType.swift:106
        for i in _parent.sources.startIndex ..< _parent.sources.endIndex {
Zip+CollectionType.swift:106
        for i in _parent.sources.startIndex ..< _parent.sources.endIndex {
Zip+CollectionType.swift:108
            let source = _parent.sources[i].asObservable()
: Parent
0017        
0018        private let _lock
Zip+CollectionType.swift:42
        _lock.lock(); defer { _lock.unlock() } // {
Zip+CollectionType.swift:42
        _lock.lock(); defer { _lock.unlock() } // {
 = NSRecursiveLock()
0019        
0020        // state
0021        private var _numberOfValues
Zip+CollectionType.swift:48
                    _numberOfValues += 1
Zip+CollectionType.swift:51
                if _numberOfValues < _parent.count {
Zip+CollectionType.swift:65
                    _numberOfValues = 0
Zip+CollectionType.swift:70
                            _numberOfValues += 1
 = 0
0022        private var _values
Zip+CollectionType.swift:29
        _values = [Queue<SourceElement>](count: parent.count, repeatedValue: Queue(capacity: 4))
Zip+CollectionType.swift:45
                _values[atIndex].enqueue(element)
Zip+CollectionType.swift:47
                if _values[atIndex].count == 1 {
Zip+CollectionType.swift:67
                    for i in 0 ..< _values.count {
Zip+CollectionType.swift:68
                        arguments.append(_values[i].dequeue()!)
Zip+CollectionType.swift:69
                        if _values[i].count > 0 {
: [Queue<SourceElement>]
0023        private var _isDone
Zip+CollectionType.swift:30
        _isDone = [Bool](count: parent.count, repeatedValue: false)
Zip+CollectionType.swift:52
                    let numberOfOthersThatAreDone = _numberOfDone - (_isDone[atIndex] ? 1 : 0)
Zip+CollectionType.swift:86
                if _isDone[atIndex] {
Zip+CollectionType.swift:90
                _isDone[atIndex] = true
: [Bool]
0024        private var _numberOfDone
Zip+CollectionType.swift:52
                    let numberOfOthersThatAreDone = _numberOfDone - (_isDone[atIndex] ? 1 : 0)
Zip+CollectionType.swift:91
                _numberOfDone += 1
Zip+CollectionType.swift:93
                if _numberOfDone == _parent.count {
 = 0
0025        private var _subscriptions
Zip+CollectionType.swift:31
        _subscriptions = Array<SingleAssignmentDisposable>()
Zip+CollectionType.swift:32
        _subscriptions.reserveCapacity(parent.count)
Zip+CollectionType.swift:35
            _subscriptions.append(SingleAssignmentDisposable())
Zip+CollectionType.swift:98
                    _subscriptions[atIndex].dispose()
Zip+CollectionType.swift:109
            _subscriptions[j].disposable = source.subscribe(AnyObserver { event in
Zip+CollectionType.swift:115
        return CompositeDisposable(disposables: _subscriptions.map { $0 })
: [SingleAssignmentDisposable]
0026        
0027        init
Zip+CollectionType.swift:133
        let sink = ZipCollectionTypeSink(parent: self, observer: observer)
(parent: Parent, observer: O) {
0028            _parent = parent
0029            _values = [Queue<SourceElement>](count: parent.count, repeatedValue: Queue(capacity: 4))
0030            _isDone = [Bool](count: parent.count, repeatedValue: false)
0031            _subscriptions = Array<SingleAssignmentDisposable>()
0032            _subscriptions.reserveCapacity(parent.count)
0033            
0034            for _ in 0 ..< parent.count {
0035                _subscriptions.append(SingleAssignmentDisposable())
0036            }
0037            
0038            super.init(observer: observer)
0039        }
0040        
0041        func on
Zip+CollectionType.swift:110
                self.on(event, atIndex: index)
(event: Event<SourceElement>, atIndex: Int) {
0042            _lock.lock(); defer { _lock.unlock() } // {
0043                switch event {
0044                case .Next(let element):
0045                    _values[atIndex].enqueue(element)
0046                    
0047                    if _values[atIndex].count == 1 {
0048                        _numberOfValues += 1
0049                    }
0050                    
0051                    if _numberOfValues < _parent.count {
0052                        let numberOfOthersThatAreDone = _numberOfDone - (_isDone[atIndex] ? 1 : 0)
0053                        if numberOfOthersThatAreDone == _parent.count - 1 {
0054                            self.forwardOn(.Completed)
0055                            self.dispose()
0056                        }
0057                        return
0058                    }
0059                    
0060                    do {
0061                        var arguments = [SourceElement]()
0062                        arguments.reserveCapacity(_parent.count)
0063                        
0064                        // recalculate number of values
0065                        _numberOfValues = 0
0066                        
0067                        for i in 0 ..< _values.count {
0068                            arguments.append(_values[i].dequeue()!)
0069                            if _values[i].count > 0 {
0070                                _numberOfValues += 1
0071                            }
0072                        }
0073                        
0074                        let result = try _parent.resultSelector(arguments)
0075                        self.forwardOn(.Next(result))
0076                    }
0077                    catch let error {
0078                        self.forwardOn(.Error(error))
0079                        self.dispose()
0080                    }
0081                    
0082                case .Error(let error):
0083                    self.forwardOn(.Error(error))
0084                    self.dispose()
0085                case .Completed:
0086                    if _isDone[atIndex] {
0087                        return
0088                    }
0089                    
0090                    _isDone[atIndex] = true
0091                    _numberOfDone += 1
0092                    
0093                    if _numberOfDone == _parent.count {
0094                        self.forwardOn(.Completed)
0095                        self.dispose()
0096                    }
0097                    else {
0098                        _subscriptions[atIndex].dispose()
0099                    }
0100                }
0101            // }
0102        }
0103        
0104        func run
Zip+CollectionType.swift:134
        sink.disposable = sink.run()
() -> Disposable {
0105            var j = 0
0106            for i in _parent.sources.startIndex ..< _parent.sources.endIndex {
0107                let index = j
0108                let source = _parent.sources[i].asObservable()
0109                _subscriptions[j].disposable = source.subscribe(AnyObserver { event in
0110                    self.on(event, atIndex: index)
0111                    })
0112                j += 1
0113            }
0114            
0115            return CompositeDisposable(disposables: _subscriptions.map { $0 })
0116        }
0117    }
0118    
0119    class ZipCollectionType
Observable+Multiple.swift:43
        return ZipCollectionType(sources: self, resultSelector: resultSelector)
Zip+CollectionType.swift:13
    typealias Parent = ZipCollectionType<C, R>
<C
Zip+CollectionType.swift:120
    typealias ResultSelector = [C.Generator.Element.E] throws -> R
Zip+CollectionType.swift:122
    let sources: C
Zip+CollectionType.swift:126
    init(sources: C, resultSelector: ResultSelector) {
: CollectionType, R
Zip+CollectionType.swift:119
class ZipCollectionType<C: CollectionType, R where C.Generator.Element : ObservableConvertibleType> : Producer<R> {
Zip+CollectionType.swift:120
    typealias ResultSelector = [C.Generator.Element.E] throws -> R
Zip+CollectionType.swift:132
    override func run<O : ObserverType where O.E == R>(observer: O) -> Disposable {
 where C.Generator.Element : ObservableConvertibleType> : Producer<R> {
0120        typealias ResultSelector
Zip+CollectionType.swift:123
    let resultSelector: ResultSelector
Zip+CollectionType.swift:126
    init(sources: C, resultSelector: ResultSelector) {
 = [C.Generator.Element.E] throws -> R
0121        
0122        let sources
Zip+CollectionType.swift:106
        for i in _parent.sources.startIndex ..< _parent.sources.endIndex {
Zip+CollectionType.swift:106
        for i in _parent.sources.startIndex ..< _parent.sources.endIndex {
Zip+CollectionType.swift:108
            let source = _parent.sources[i].asObservable()
Zip+CollectionType.swift:127
        self.sources = sources
Zip+CollectionType.swift:129
        self.count = Int(self.sources.count.toIntMax())
: C
0123        let resultSelector
Zip+CollectionType.swift:74
                    let result = try _parent.resultSelector(arguments)
Zip+CollectionType.swift:128
        self.resultSelector = resultSelector
: ResultSelector
0124        let count
Zip+CollectionType.swift:29
        _values = [Queue<SourceElement>](count: parent.count, repeatedValue: Queue(capacity: 4))
Zip+CollectionType.swift:30
        _isDone = [Bool](count: parent.count, repeatedValue: false)
Zip+CollectionType.swift:32
        _subscriptions.reserveCapacity(parent.count)
Zip+CollectionType.swift:34
        for _ in 0 ..< parent.count {
Zip+CollectionType.swift:51
                if _numberOfValues < _parent.count {
Zip+CollectionType.swift:53
                    if numberOfOthersThatAreDone == _parent.count - 1 {
Zip+CollectionType.swift:62
                    arguments.reserveCapacity(_parent.count)
Zip+CollectionType.swift:93
                if _numberOfDone == _parent.count {
Zip+CollectionType.swift:129
        self.count = Int(self.sources.count.toIntMax())
: Int
0125        
0126        init
Observable+Multiple.swift:43
        return ZipCollectionType(sources: self, resultSelector: resultSelector)
(sources: C, resultSelector: ResultSelector) {
0127            self.sources = sources
0128            self.resultSelector = resultSelector
0129            self.count = Int(self.sources.count.toIntMax())
0130        }
0131        
0132        override func run<O : ObserverType where O.E == R>(observer: O) -> Disposable {
0133            let sink = ZipCollectionTypeSink(parent: self, observer: observer)
0134            sink.disposable = sink.run()
0135            return sink
0136        }
0137    }
0138