0001    //
0002    //  Throttle.swift
0003    //  Rx
0004    //
0005    //  Created by Krunoslav Zaher on 3/22/15.
0006    //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
0007    //
0008    
0009    import Foundation
0010    
0011    class ThrottleSink
Throttle.swift:99
        let sink = ThrottleSink(parent: self, observer: observer)
<O
Throttle.swift:12
    : Sink<O>
Throttle.swift:16
    typealias Element = O.E
Throttle.swift:29
    init(parent: ParentType, observer: O) {
: ObserverType>
0012        : Sink<O>
0013        , ObserverType
0014        , LockOwnerType
0015        , SynchronizedOnType {
0016        typealias Element
Throttle.swift:17
    typealias ParentType = Throttle<Element>
Throttle.swift:25
    private var _value: Element? = nil
Throttle.swift:41
    func on(event: Event<Element>) {
Throttle.swift:45
    func _synchronized_on(event: Event<Element>) {
 = O.E
0017        typealias ParentType
Throttle.swift:19
    private let _parent: ParentType
Throttle.swift:29
    init(parent: ParentType, observer: O) {
 = Throttle<Element>
0018        
0019        private let _parent
Throttle.swift:30
        _parent = parent
Throttle.swift:36
        let subscription = _parent._source.subscribe(self)
Throttle.swift:53
            let scheduler = _parent._scheduler
Throttle.swift:54
            let dueTime = _parent._dueTime
: ParentType
0020        
0021        let _lock
Throttle.swift:74
        _lock.lock(); defer { _lock.unlock() } // {
Throttle.swift:74
        _lock.lock(); defer { _lock.unlock() } // {
 = NSRecursiveLock()
0022        
0023        // state
0024        private var _id
Throttle.swift:48
            _id = _id &+ 1
Throttle.swift:48
            _id = _id &+ 1
Throttle.swift:49
            let currentId = _id
Throttle.swift:77
            if let value = originalValue where _id == currentId {
 = 0 as UInt64
0025        private var _value
Throttle.swift:50
            _value = element
Throttle.swift:60
            _value = nil
Throttle.swift:64
            if let value = _value {
Throttle.swift:65
                _value = nil
Throttle.swift:75
            let originalValue = _value
Throttle.swift:78
                _value = nil
: Element? = nil
0026        
0027        let cancellable
Throttle.swift:38
        return StableCompositeDisposable.create(subscription, cancellable)
Throttle.swift:57
            self.cancellable.disposable = d
 = SerialDisposable()
0028        
0029        init
Throttle.swift:99
        let sink = ThrottleSink(parent: self, observer: observer)
(parent: ParentType, observer: O) {
0030            _parent = parent
0031            
0032            super.init(observer: observer)
0033        }
0034        
0035        func run
Throttle.swift:100
        sink.disposable = sink.run()
() -> Disposable {
0036            let subscription = _parent._source.subscribe(self)
0037            
0038            return StableCompositeDisposable.create(subscription, cancellable)
0039        }
0040    
0041        func on(event: Event<Element>) {
0042            synchronizedOn(event)
0043        }
0044    
0045        func _synchronized_on(event: Event<Element>) {
0046            switch event {
0047            case .Next(let element):
0048                _id = _id &+ 1
0049                let currentId = _id
0050                _value = element
0051    
0052                
0053                let scheduler = _parent._scheduler
0054                let dueTime = _parent._dueTime
0055    
0056                let d = SingleAssignmentDisposable()
0057                self.cancellable.disposable = d
0058                d.disposable = scheduler.scheduleRelative(currentId, dueTime: dueTime, action: self.propagate)
0059            case .Error:
0060                _value = nil
0061                forwardOn(event)
0062                dispose()
0063            case .Completed:
0064                if let value = _value {
0065                    _value = nil
0066                    forwardOn(.Next(value))
0067                }
0068                forwardOn(.Completed)
0069                dispose()
0070            }
0071        }
0072        
0073        func propagate
Throttle.swift:58
            d.disposable = scheduler.scheduleRelative(currentId, dueTime: dueTime, action: self.propagate)
(currentId: UInt64) -> Disposable {
0074            _lock.lock(); defer { _lock.unlock() } // {
0075                let originalValue = _value
0076    
0077                if let value = originalValue where _id == currentId {
0078                    _value = nil
0079                    forwardOn(.Next(value))
0080                }
0081            // }
0082            return NopDisposable.instance
0083        }
0084    }
0085    
0086    class Throttle
Observable+Time.swift:28
        return Throttle(source: self.asObservable(), dueTime: dueTime, scheduler: scheduler)
Observable+Time.swift:45
        return Throttle(source: self.asObservable(), dueTime: dueTime, scheduler: scheduler)
Throttle.swift:17
    typealias ParentType = Throttle<Element>
<Element
Throttle.swift:86
class Throttle<Element> : Producer<Element> {
Throttle.swift:88
    private let _source: Observable<Element>
Throttle.swift:92
    init(source: Observable<Element>, dueTime: RxTimeInterval, scheduler: SchedulerType) {
Throttle.swift:98
    override func run<O: ObserverType where O.E == Element>(observer: O) -> Disposable {
> : Producer<Element> {
0087        
0088        private let _source
Throttle.swift:36
        let subscription = _parent._source.subscribe(self)
Throttle.swift:93
        _source = source
: Observable<Element>
0089        private let _dueTime
Throttle.swift:54
            let dueTime = _parent._dueTime
Throttle.swift:94
        _dueTime = dueTime
: RxTimeInterval
0090        private let _scheduler
Throttle.swift:53
            let scheduler = _parent._scheduler
Throttle.swift:95
        _scheduler = scheduler
: SchedulerType
0091        
0092        init
Observable+Time.swift:28
        return Throttle(source: self.asObservable(), dueTime: dueTime, scheduler: scheduler)
Observable+Time.swift:45
        return Throttle(source: self.asObservable(), dueTime: dueTime, scheduler: scheduler)
(source: Observable<Element>, dueTime: RxTimeInterval, scheduler: SchedulerType) {
0093            _source = source
0094            _dueTime = dueTime
0095            _scheduler = scheduler
0096        }
0097        
0098        override func run<O: ObserverType where O.E == Element>(observer: O) -> Disposable {
0099            let sink = ThrottleSink(parent: self, observer: observer)
0100            sink.disposable = sink.run()
0101            return sink
0102        }
0103        
0104    }