0001    //
0002    //  ConcurrentDispatchQueueScheduler.swift
0003    //  RxSwift
0004    //
0005    //  Created by Krunoslav Zaher on 7/5/15.
0006    //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
0007    //
0008    
0009    import Foundation
0010    
0011    /**
0012    Abstracts the work that needs to be performed on a specific `dispatch_queue_t`. You can also pass a serial dispatch queue, it shouldn't cause any problems.
0013    
0014    This scheduler is suitable when some work needs to be performed in background.
0015    */
0016    public class ConcurrentDispatchQueueScheduler: SchedulerType {
0017        public typealias TimeInterval
ConcurrentDispatchQueueScheduler.swift:125
    public func schedulePeriodic<StateType>(state: StateType, startAfter: TimeInterval, period: TimeInterval, action: (StateType) -> StateType) -> Disposable {
ConcurrentDispatchQueueScheduler.swift:125
    public func schedulePeriodic<StateType>(state: StateType, startAfter: TimeInterval, period: TimeInterval, action: (StateType) -> StateType) -> Disposable { = NSTimeInterval
0018        public typealias Time = NSDate
0019        
0020        private let _queue
ConcurrentDispatchQueueScheduler.swift:37
        _queue = queue
ConcurrentDispatchQueueScheduler.swift:74
        dispatch_async(_queue) {
ConcurrentDispatchQueueScheduler.swift:94
        let timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, _queue)
ConcurrentDispatchQueueScheduler.swift:126
        let timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, _queue) : dispatch_queue_t
0021        
0022        public var now : NSDate {
0023            get {
0024                return NSDate()
0025            }
0026        }
0027        
0028        // leeway for scheduling timers
0029        private var _leeway: Int64 = 
0030        
0031        /**
0032        Constructs new `ConcurrentDispatchQueueScheduler` that wraps `queue`.
0033        
0034        - parameter queue: Target dispatch queue.
0035        */
0036        public init
ConcurrentDispatchQueueScheduler.swift:48
        self.init(queue: dispatch_get_global_queue(priority, UInt(0)))(queue: dispatch_queue_t) {
0037            _queue = queue
0038        }
0039        
0040        /**
0041         Convenience init for scheduler that wraps one of the global concurrent dispatch queues.
0042         
0043         - parameter globalConcurrentQueueQOS: Target global dispatch queue, by quality of service class.
0044         */
0045        @available(iOS , OSX 10.10, *)
0046        public convenience init(globalConcurrentQueueQOS: DispatchQueueSchedulerQOS) {
0047            let priority = globalConcurrentQueueQOS.QOSClass
0048            self.init(queue: dispatch_get_global_queue(priority, UInt()))
0049        }
0050    
0051        
0052        class func convertTimeIntervalToDispatchInterval
ConcurrentDispatchQueueScheduler.swift:57
        return dispatch_time(DISPATCH_TIME_NOW, convertTimeIntervalToDispatchInterval(timeInterval))(timeInterval: NSTimeInterval) -> Int64 {
0053            return Int64(timeInterval * Double(NSEC_PER_SEC))
0054        }
0055        
0056        class func convertTimeIntervalToDispatchTime(timeInterval: NSTimeInterval) -> dispatch_time_t {
0057            return dispatch_time(DISPATCH_TIME_NOW, convertTimeIntervalToDispatchInterval(timeInterval))
0058        }
0059        
0060        /**
0061        Schedules an action to be executed immediatelly.
0062        
0063        - parameter state: State passed to the action to be executed.
0064        - parameter action: Action to be executed.
0065        - returns: The disposable object used to cancel the scheduled action (best effort).
0066        */
0067        public final func schedule<StateType>(state: StateType, action: StateType -> Disposable) -> Disposable {
0068            return self.scheduleInternal(state, action: action)
0069        }
0070        
0071        func scheduleInternal
ConcurrentDispatchQueueScheduler.swift:68
        return self.scheduleInternal(state, action: action)<StateType>(state: StateType, action: StateType -> Disposable) -> Disposable {
0072            let cancel = SingleAssignmentDisposable()
0073            
0074            dispatch_async(_queue) {
0075                if cancel.disposed {
0076                    return
0077                }
0078                
0079                cancel.disposable = action(state)
0080            }
0081            
0082            return cancel
0083        }
0084        
0085        /**
0086        Schedules an action to be executed.
0087        
0088        - parameter state: State passed to the action to be executed.
0089        - parameter dueTime: Relative time after which to execute the action.
0090        - parameter action: Action to be executed.
0091        - returns: The disposable object used to cancel the scheduled action (best effort).
0092        */
0093        public final func scheduleRelative<StateType>(state: StateType, dueTime: NSTimeInterval, action: (StateType) -> Disposable) -> Disposable {
0094            let timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, , , _queue)
0095            
0096            let dispatchInterval = MainScheduler.convertTimeIntervalToDispatchTime(dueTime)
0097            
0098            let compositeDisposable = CompositeDisposable()
0099            
0100            dispatch_source_set_timer(timer, dispatchInterval, DISPATCH_TIME_FOREVER, )
0101            dispatch_source_set_event_handler(timer, {
0102                if compositeDisposable.disposed {
0103                    return
0104                }
0105               compositeDisposable.addDisposable(action(state))
0106            })
0107            dispatch_resume(timer)
0108            
0109            compositeDisposable.addDisposable(AnonymousDisposable {
0110                dispatch_source_cancel(timer)
0111                })
0112            
0113            return compositeDisposable
0114        }
0115        
0116        /**
0117        Schedules a periodic piece of work.
0118        
0119        - parameter state: State passed to the action to be executed.
0120        - parameter startAfter: Period after which initial work should be run.
0121        - parameter period: Period for running the work periodically.
0122        - parameter action: Action to be executed.
0123        - returns: The disposable object used to cancel the scheduled action (best effort).
0124        */
0125        public func schedulePeriodic<StateType>(state: StateType, startAfter: TimeInterval, period: TimeInterval, action: (StateType) -> StateType) -> Disposable {
0126            let timer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, , , _queue)
0127            
0128            let initial = MainScheduler.convertTimeIntervalToDispatchTime(startAfter)
0129            let dispatchInterval = MainScheduler.convertTimeIntervalToDispatchInterval(period)
0130            
0131            var timerState = state
0132            
0133            let validDispatchInterval = dispatchInterval <  ?  : UInt64(dispatchInterval)
0134            
0135            dispatch_source_set_timer(timer, initial, validDispatchInterval, )
0136            let cancel = AnonymousDisposable {
0137                dispatch_source_cancel(timer)
0138            }
0139            dispatch_source_set_event_handler(timer, {
0140                if cancel.disposed {
0141                    return
0142                }
0143                timerState = action(timerState)
0144            })
0145            dispatch_resume(timer)
0146            
0147            return cancel
0148        }
0149    }