RxSwift-article-8.swift

RxSwift-article-8.swift

public override func subscribe<O : ObserverType where O.E == Element>(observer: O) -> Disposable {
    _lock.lock(); defer { _lock.unlock() }
    return _synchronized_subscribe(observer)
}

func _synchronized_subscribe<O : ObserverType where O.E == E>(observer: O) -> Disposable {
    if _disposed {
        observer.on(.Error(RxError.Disposed(object: self)))
        return NopDisposable.instance
    }
    
    if let stoppedEvent = _stoppedEvent {
        observer.on(stoppedEvent)
        return NopDisposable.instance
    }
    
    let key = _observers.insert(observer.asObserver())
    observer.on(.Next(_value))

    return SubscriptionDisposable(owner: self, key: key)
}