PromiseObservableConversion.swift

//
//  Observable+Promise.swift
//
//  Created by Charles Julian Knight on 1/2/17.
//  Copyright © 2017 Charles Julian Knight. All rights reserved.
//

import Foundation
import RxSwift
import PromiseKit

extension ObservableType {
    
    /// Convert an observable into a promise.
    ///
    /// Warning! This will cause the observable
    /// to immediately begin executing. It will
    /// also error if the observable returns more
    /// than one result. The observable will
    /// also not be cancelable, and resources will
    func toPromise() -> Promise<Self.E> {
        return ObserablePromise(self.asObservable()).promise
    }
}

fileprivate class ObserablePromise<T>: ObserverType {
    typealias E = T
    
    let observable: Observable<T>
    var disposable: Disposable?
    
    let (_promise, fulfilled, rejected) = Promise<T>.pending()
    
    init(_ observable: Observable<T>) {
        self.observable = observable
    }

    fileprivate func on(_ event: Event<T>) {
        switch event {
        case .next(let t):
            if _promise.isPending {
                fulfilled(t)
                disposable?.dispose()
                disposable = nil
            }else{
                fatalError("Promise already completed")
            }
        case .completed:
            if !_promise.isPending {
                fatalError("No items emitted")
            }
        case .error(let error):
            if _promise.isPending {
                rejected(error)
                disposable?.dispose()
                disposable = nil
            }else{
                fatalError("Promise already completed")
            }
        }
    }
    
    var promise: Promise<T> {
        guard disposable == nil else {
            fatalError("Promise already created")
        }
        self.disposable = observable.single().subscribe(self)
        return _promise
    }
}