//
//  Delay.swift
//  RxSwift
//
//  Created by tarunon on 2016/02/09.
//  Copyright © 2016 Krunoslav Zaher. All rights reserved.
//

import Foundation

extension ObservableType {

    /**
     Returns an observable sequence by the source observable sequence shifted forward in time by a specified delay. Error events from the source observable sequence are not delayed.

     - seealso: [delay operator on reactivex.io](http://reactivex.io/documentation/operators/delay.html)

     - parameter dueTime: Relative time shift of the source by.
     - parameter scheduler: Scheduler to run the subscription delay timer on.
     - returns: the source Observable shifted in time by the specified delay.
     */
    public func delay(_ dueTime: RxTimeInterval, scheduler: SchedulerType)
        -> Observable<Element> {
            return Delay(source: self.asObservable(), dueTime: dueTime, scheduler: scheduler)
    }
}

final private class DelaySink<Observer: ObserverType>
    : Sink<Observer>
    , ObserverType {
    typealias Element = Observer.Element 
    typealias Source = Observable<Element>
    typealias DisposeKey = Bag<Disposable>.KeyType
    
    private let lock = RecursiveLock()

    private let dueTime: RxTimeInterval
    private let scheduler: SchedulerType
    
    private let sourceSubscription = SingleAssignmentDisposable()
    private let cancelable = SerialDisposable()

    // is scheduled some action
    private var active = false
    // is "run loop" on different scheduler running
    private var running = false
    private var errorEvent: Event<Element>?

    // state
    private var queue = Queue<(eventTime: RxTime, event: Event<Element>)>(capacity: 0)
    
    init(observer: Observer, dueTime: RxTimeInterval, scheduler: SchedulerType, cancel: Cancelable) {
        self.dueTime = dueTime
        self.scheduler = scheduler
        super.init(observer: observer, cancel: cancel)
    }

    // All of these complications in this method are caused by the fact that 
    // error should be propagated immediately. Error can be potentially received on different
    // scheduler so this process needs to be synchronized somehow.
    //
    // Another complication is that scheduler is potentially concurrent so internal queue is used.
    func drainQueue(state: (), scheduler: AnyRecursiveScheduler<()>) {
        self.lock.lock()    
        let hasFailed = self.errorEvent != nil
        if !hasFailed {
            self.running = true
        }
        self.lock.unlock()  

        if hasFailed {
            return
        }

        var ranAtLeastOnce = false

        while true {
            self.lock.lock() 
            let errorEvent = self.errorEvent

            let eventToForwardImmediately = ranAtLeastOnce ? nil : self.queue.dequeue()?.event
            let nextEventToScheduleOriginalTime: Date? = ranAtLeastOnce && !self.queue.isEmpty ? self.queue.peek().eventTime : nil

            if errorEvent == nil {
                if eventToForwardImmediately != nil {
                }
                else if nextEventToScheduleOriginalTime != nil {
                    self.running = false
                }
                else {
                    self.running = false
                    self.active = false
                }
            }
            self.lock.unlock() 

            if let errorEvent = errorEvent {
                self.forwardOn(errorEvent)
                self.dispose()
                return
            }
            else {
                if let eventToForwardImmediately = eventToForwardImmediately {
                    ranAtLeastOnce = true
                    self.forwardOn(eventToForwardImmediately)
                    if case .completed = eventToForwardImmediately {
                        self.dispose()
                        return
                    }
                }
                else if let nextEventToScheduleOriginalTime = nextEventToScheduleOriginalTime {
                    scheduler.schedule((), dueTime: self.dueTime.reduceWithSpanBetween(earlierDate: nextEventToScheduleOriginalTime, laterDate: self.scheduler.now))
                    return
                }
                else {
                    return
                }
            }
        }
    }
    
    func on(_ event: Event<Element>) {
        if event.isStopEvent {
            self.sourceSubscription.dispose()
        }

        switch event {
        case .error:
            self.lock.lock()    
            let shouldSendImmediately = !self.running
            self.queue = Queue(capacity: 0)
            self.errorEvent = event
            self.lock.unlock()  

            if shouldSendImmediately {
                self.forwardOn(event)
                self.dispose()
            }
        default:
            self.lock.lock()    
            let shouldSchedule = !self.active
            self.active = true
            self.queue.enqueue((self.scheduler.now, event))
            self.lock.unlock()  

            if shouldSchedule {
                self.cancelable.disposable = self.scheduler.scheduleRecursive((), dueTime: self.dueTime, action: self.drainQueue)
            }
        }
    }
    
    func run(source: Observable<Element>) -> Disposable {
        self.sourceSubscription.setDisposable(source.subscribe(self))
        return Disposables.create(sourceSubscription, cancelable)
    }
}

final private class Delay<Element>: Producer<Element> {
    private let source: Observable<Element>
    private let dueTime: RxTimeInterval
    private let scheduler: SchedulerType
    
    init(source: Observable<Element>, dueTime: RxTimeInterval, scheduler: SchedulerType) {
        self.source = source
        self.dueTime = dueTime
        self.scheduler = scheduler
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        let sink = DelaySink(observer: observer, dueTime: self.dueTime, scheduler: self.scheduler, cancel: cancel)
        let subscription = sink.run(source: self.source)
        return (sink: sink, subscription: subscription)
    }
}