0001    //
0002    //  Timeout.swift
0003    //  Rx
0004    //
0005    //  Created by Tomi Koskinen on 13/11/15.
0006    //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
0007    //
0008    
0009    import Foundation
0010    
0011    class TimeoutSink
Timeout.swift:116
        let sink = TimeoutSink(parent: self, observer: observer)
<ElementType
Timeout.swift:12
    typealias E = ElementType
, O
Timeout.swift:11
class TimeoutSink<ElementType, O: ObserverType where O.E == ElementType>: Sink<O>, LockOwnerType, ObserverType {
Timeout.swift:25
    init(parent: Parent, observer: O) {
: ObserverType where O.E == ElementType>: Sink<O>, LockOwnerType, ObserverType {
0012        typealias E
Timeout.swift:13
    typealias Parent = Timeout<E>
Timeout.swift:41
    func on(event: Event<E>) {
 = ElementType
0013        typealias Parent
Timeout.swift:15
    private let _parent: Parent
Timeout.swift:25
    init(parent: Parent, observer: O) {
 = Timeout<E>
0014        
0015        private let _parent
Timeout.swift:26
        _parent = parent
Timeout.swift:36
        original.disposable = _parent._source.subscribeSafe(self)
Timeout.swift:82
        nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
Timeout.swift:82
        nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
Timeout.swift:92
                self._subscription.disposable = self._parent._other.subscribeSafe(self.forwarder())
: Parent
0016        
0017        let _lock
Timeout.swift:46
            _lock.performLocked() {
Timeout.swift:60
            _lock.performLocked() {
Timeout.swift:86
            self._lock.performLocked() {
 = NSRecursiveLock()
0018    
0019        private let _timerD
Timeout.swift:38
        return StableCompositeDisposable.create(_subscription, _timerD)
Timeout.swift:75
        if _timerD.disposed {
Timeout.swift:80
        _timerD.disposable = nextTimer
 = SerialDisposable()
0020        private let _subscription
Timeout.swift:32
        _subscription.disposable = original
Timeout.swift:38
        return StableCompositeDisposable.create(_subscription, _timerD)
Timeout.swift:92
                self._subscription.disposable = self._parent._other.subscribeSafe(self.forwarder())
 = SerialDisposable()
0021        
0022        private var _id
Timeout.swift:49
                    self._id = self._id &+ 1
Timeout.swift:49
                    self._id = self._id &+ 1
Timeout.swift:63
                    self._id = self._id &+ 1
Timeout.swift:63
                    self._id = self._id &+ 1
Timeout.swift:82
        nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
Timeout.swift:87
                self._switched = (state == self._id)
 = 0
0023        private var _switched
Timeout.swift:47
                onNextWins = !self._switched
Timeout.swift:61
                onEventWins = !self._switched
Timeout.swift:87
                self._switched = (state == self._id)
Timeout.swift:88
                timerWins = self._switched
 = false
0024        
0025        init
Timeout.swift:116
        let sink = TimeoutSink(parent: self, observer: observer)
(parent: Parent, observer: O) {
0026            _parent = parent
0027            super.init(observer: observer)
0028        }
0029        
0030        func run
Timeout.swift:117
        sink.disposable = sink.run()
() -> Disposable {
0031            let original = SingleAssignmentDisposable()
0032            _subscription.disposable = original
0033            
0034            _createTimeoutTimer()
0035            
0036            original.disposable = _parent._source.subscribeSafe(self)
0037            
0038            return StableCompositeDisposable.create(_subscription, _timerD)
0039        }
0040    
0041        func on(event: Event<E>) {
0042            switch event {
0043            case .Next:
0044                var onNextWins = false
0045                
0046                _lock.performLocked() {
0047                    onNextWins = !self._switched
0048                    if onNextWins {
0049                        self._id = self._id &+ 1
0050                    }
0051                }
0052                
0053                if onNextWins {
0054                    forwardOn(event)
0055                    self._createTimeoutTimer()
0056                }
0057            case .Error, .Completed:
0058                var onEventWins = false
0059                
0060                _lock.performLocked() {
0061                    onEventWins = !self._switched
0062                    if onEventWins {
0063                        self._id = self._id &+ 1
0064                    }
0065                }
0066                
0067                if onEventWins {
0068                    forwardOn(event)
0069                    self.dispose()
0070                }
0071            }
0072        }
0073        
0074        private func _createTimeoutTimer
Timeout.swift:34
        _createTimeoutTimer()
Timeout.swift:55
                self._createTimeoutTimer()
() {
0075            if _timerD.disposed {
0076                return
0077            }
0078            
0079            let nextTimer = SingleAssignmentDisposable()
0080            _timerD.disposable = nextTimer
0081            
0082            nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
0083                
0084                var timerWins = false
0085                
0086                self._lock.performLocked() {
0087                    self._switched = (state == self._id)
0088                    timerWins = self._switched
0089                }
0090                
0091                if timerWins {
0092                    self._subscription.disposable = self._parent._other.subscribeSafe(self.forwarder())
0093                }
0094                
0095                return NopDisposable.instance
0096            }
0097        }
0098    }
0099    
0100    
0101    class Timeout
Observable+Time.swift:256
            return Timeout(source: self.asObservable(), dueTime: dueTime, other: Observable.error(RxError.Timeout), scheduler: scheduler)
Observable+Time.swift:272
            return Timeout(source: self.asObservable(), dueTime: dueTime, other: other.asObservable(), scheduler: scheduler)
Timeout.swift:13
    typealias Parent = Timeout<E>
<Element
Timeout.swift:101
class Timeout<Element> : Producer<Element> {
Timeout.swift:103
    private let _source: Observable<Element>
Timeout.swift:105
    private let _other: Observable<Element>
Timeout.swift:108
    init(source: Observable<Element>, dueTime: RxTimeInterval, other: Observable<Element>, scheduler: SchedulerType) {
Timeout.swift:108
    init(source: Observable<Element>, dueTime: RxTimeInterval, other: Observable<Element>, scheduler: SchedulerType) {
Timeout.swift:115
    override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
> : Producer<Element> {
0102        
0103        private let _source
Timeout.swift:36
        original.disposable = _parent._source.subscribeSafe(self)
Timeout.swift:109
        _source = source
: Observable<Element>
0104        private let _dueTime
Timeout.swift:82
        nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
Timeout.swift:110
        _dueTime = dueTime
: RxTimeInterval
0105        private let _other
Timeout.swift:92
                self._subscription.disposable = self._parent._other.subscribeSafe(self.forwarder())
Timeout.swift:111
        _other = other
: Observable<Element>
0106        private let _scheduler
Timeout.swift:82
        nextTimer.disposable = _parent._scheduler.scheduleRelative(_id, dueTime: _parent._dueTime) { state in
Timeout.swift:112
        _scheduler = scheduler
: SchedulerType
0107        
0108        init
Observable+Time.swift:256
            return Timeout(source: self.asObservable(), dueTime: dueTime, other: Observable.error(RxError.Timeout), scheduler: scheduler)
Observable+Time.swift:272
            return Timeout(source: self.asObservable(), dueTime: dueTime, other: other.asObservable(), scheduler: scheduler)
(source: Observable<Element>, dueTime: RxTimeInterval, other: Observable<Element>, scheduler: SchedulerType) {
0109            _source = source
0110            _dueTime = dueTime
0111            _other = other
0112            _scheduler = scheduler
0113        }
0114        
0115        override func run<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
0116            let sink = TimeoutSink(parent: self, observer: observer)
0117            sink.disposable = sink.run()
0118            return sink
0119        }
0120    }
0121