0001    //
0002    //  Take.swift
0003    //  RxSwift
0004    //
0005    //  Created by Krunoslav Zaher on 6/12/15.
0006    //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
0007    //
0008    
0009    import Foundation
0010    
0011    // count version
0012    
0013    class TakeCountSink
Take.swift:65
        let sink = TakeCountSink(parent: self, observer: observer)
<ElementType
Take.swift:14
    typealias Parent = TakeCount<ElementType>
Take.swift:15
    typealias E = ElementType
, O
Take.swift:13
class TakeCountSink<ElementType, O: ObserverType where O.E == ElementType> : Sink<O>, ObserverType {
Take.swift:21
    init(parent: Parent, observer: O) {
: ObserverType where O.E == ElementType> : Sink<O>, ObserverType {
0014        typealias Parent
Take.swift:17
    private let _parent: Parent
Take.swift:21
    init(parent: Parent, observer: O) {
 = TakeCount<ElementType>
0015        typealias E
Take.swift:27
    func on(event: Event<E>) {
 = ElementType
0016        
0017        private let _parent
Take.swift:22
        _parent = parent
: Parent
0018        
0019        private var _remaining
Take.swift:23
        _remaining = parent._count
Take.swift:31
            if _remaining > 0 {
Take.swift:32
                _remaining -= 1
Take.swift:36
                if _remaining == 0 {
: Int
0020        
0021        init
Take.swift:65
        let sink = TakeCountSink(parent: self, observer: observer)
(parent: Parent, observer: O) {
0022            _parent = parent
0023            _remaining = parent._count
0024            super.init(observer: observer)
0025        }
0026        
0027        func on(event: Event<E>) {
0028            switch event {
0029            case .Next(let value):
0030                
0031                if _remaining > 0 {
0032                    _remaining -= 1
0033                    
0034                    forwardOn(.Next(value))
0035                
0036                    if _remaining == 0 {
0037                        forwardOn(.Completed)
0038                        dispose()
0039                    }
0040                }
0041            case .Error:
0042                forwardOn(event)
0043                dispose()
0044            case .Completed:
0045                forwardOn(event)
0046                dispose()
0047            }
0048        }
0049        
0050    }
0051    
0052    class TakeCount
Observable+StandardSequenceOperators.swift:84
            return TakeCount(source: asObservable(), count: count)
Take.swift:14
    typealias Parent = TakeCount<ElementType>
<Element
Take.swift:52
class TakeCount<Element>: Producer<Element> {
Take.swift:53
    private let _source: Observable<Element>
Take.swift:56
    init(source: Observable<Element>, count: Int) {
Take.swift:64
    override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
>: Producer<Element> {
0053        private let _source
Take.swift:60
        _source = source
Take.swift:66
        sink.disposable = _source.subscribe(sink)
: Observable<Element>
0054        private let _count
Take.swift:23
        _remaining = parent._count
Take.swift:61
        _count = count
: Int
0055        
0056        init
Observable+StandardSequenceOperators.swift:84
            return TakeCount(source: asObservable(), count: count)
(source: Observable<Element>, count: Int) {
0057            if count < 0 {
0058                rxFatalError("count can't be negative")
0059            }
0060            _source = source
0061            _count = count
0062        }
0063        
0064        override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
0065            let sink = TakeCountSink(parent: self, observer: observer)
0066            sink.disposable = _source.subscribe(sink)
0067            return sink
0068        }
0069    }
0070    
0071    // time version
0072    
0073    class TakeTimeSink
Take.swift:140
        let sink = TakeTimeSink(parent: self, observer: observer)
<ElementType
Take.swift:78
    typealias Parent = TakeTime<ElementType>
Take.swift:79
    typealias E = ElementType
, O
Take.swift:74
    : Sink<O>
Take.swift:85
    init(parent: Parent, observer: O) {
: ObserverType where O.E == ElementType>
0074        : Sink<O>
0075        , LockOwnerType
0076        , ObserverType
0077        , SynchronizedOnType {
0078        typealias Parent
Take.swift:81
    private let _parent: Parent
Take.swift:85
    init(parent: Parent, observer: O) {
 = TakeTime<ElementType>
0079        typealias E
Take.swift:90
    func on(event: Event<E>) {
Take.swift:94
    func _synchronized_on(event: Event<E>) {
 = ElementType
0080    
0081        private let _parent
Take.swift:86
        _parent = parent
Take.swift:115
        let disposeTimer = _parent._scheduler.scheduleRelative((), dueTime: _parent._duration) {
Take.swift:115
        let disposeTimer = _parent._scheduler.scheduleRelative((), dueTime: _parent._duration) {
Take.swift:120
        let disposeSubscription = _parent._source.subscribe(self)
: Parent
0082        
0083        let _lock
Take.swift:108
        _lock.lock(); defer { _lock.unlock() }
Take.swift:108
        _lock.lock(); defer { _lock.unlock() }
 = NSRecursiveLock()
0084        
0085        init
Take.swift:140
        let sink = TakeTimeSink(parent: self, observer: observer)
(parent: Parent, observer: O) {
0086            _parent = parent
0087            super.init(observer: observer)
0088        }
0089        
0090        func on(event: Event<E>) {
0091            synchronizedOn(event)
0092        }
0093    
0094        func _synchronized_on(event: Event<E>) {
0095            switch event {
0096            case .Next(let value):
0097                forwardOn(.Next(value))
0098            case .Error:
0099                forwardOn(event)
0100                dispose()
0101            case .Completed:
0102                forwardOn(event)
0103                dispose()
0104            }
0105        }
0106        
0107        func tick
Take.swift:116
            self.tick()
() {
0108            _lock.lock(); defer { _lock.unlock() }
0109    
0110            forwardOn(.Completed)
0111            dispose()
0112        }
0113        
0114        func run
Take.swift:141
        sink.disposable = sink.run()
() -> Disposable {
0115            let disposeTimer = _parent._scheduler.scheduleRelative((), dueTime: _parent._duration) {
0116                self.tick()
0117                return NopDisposable.instance
0118            }
0119            
0120            let disposeSubscription = _parent._source.subscribe(self)
0121            
0122            return BinaryDisposable(disposeTimer, disposeSubscription)
0123        }
0124    }
0125    
0126    class TakeTime
Observable+Time.swift:132
        return TakeTime(source: self.asObservable(), duration: duration, scheduler: scheduler)
Take.swift:78
    typealias Parent = TakeTime<ElementType>
<Element
Take.swift:126
class TakeTime<Element> : Producer<Element> {
Take.swift:129
    private let _source: Observable<Element>
Take.swift:133
    init(source: Observable<Element>, duration: TimeInterval, scheduler: SchedulerType) {
Take.swift:139
    override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
> : Producer<Element> {
0127        typealias TimeInterval
Take.swift:130
    private let _duration: TimeInterval
Take.swift:133
    init(source: Observable<Element>, duration: TimeInterval, scheduler: SchedulerType) {
 = RxTimeInterval
0128        
0129        private let _source
Take.swift:120
        let disposeSubscription = _parent._source.subscribe(self)
Take.swift:134
        _source = source
: Observable<Element>
0130        private let _duration
Take.swift:115
        let disposeTimer = _parent._scheduler.scheduleRelative((), dueTime: _parent._duration) {
Take.swift:136
        _duration = duration
: TimeInterval
0131        private let _scheduler
Take.swift:115
        let disposeTimer = _parent._scheduler.scheduleRelative((), dueTime: _parent._duration) {
Take.swift:135
        _scheduler = scheduler
: SchedulerType
0132        
0133        init
Observable+Time.swift:132
        return TakeTime(source: self.asObservable(), duration: duration, scheduler: scheduler)
(source: Observable<Element>, duration: TimeInterval, scheduler: SchedulerType) {
0134            _source = source
0135            _scheduler = scheduler
0136            _duration = duration
0137        }
0138        
0139        override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
0140            let sink = TakeTimeSink(parent: self, observer: observer)
0141            sink.disposable = sink.run()
0142            return sink
0143        }
0144    }