An operation queue for Promises.

PromiseQueue.swift

import Dispatch
import PromiseKit

/// A wrapper for a function that creates a Promise.
public class PromiseOperation<T> {
    private let makePromise: () -> Promise<T>

    public init(makePromise: @escaping () -> Promise<T>) {
        self.makePromise = makePromise
    }

    /// Run the operation.
    ///
    /// - Returns: The `Promise` returned by the operation.
    public func run() -> Promise<T> {
        return makePromise()
    }

    /// Run the operation and block until it finishes.
    ///
    /// - Returns: The `Promise` returned by the operation.
    public func runAndWait() -> Promise<T> {
        let semaphore = DispatchSemaphore(value: 0)
        let promise = makePromise().always(on: zalgo) {
            semaphore.signal()
        }
        _ = semaphore.wait(timeout: DispatchTime.distantFuture)
        return promise
    }
}

/// An operation queue that may run `PromiseOperations` concurrently and supports
/// barrier operations.
public class PromiseQueue {
    private let queue: DispatchQueue

    internal init(queue: DispatchQueue) {
        self.queue = queue
    }

    public init(name: String) {
        self.init(queue: DispatchQueue(label: name, attributes: DispatchQueue.Attributes.concurrent))
    }

    /// Submits an operation for asynchronous execution.
    ///
    /// - Parameter operation: The operation to run.
    ///
    /// - Returns: The `Promise` returned by running `operation`.
    public func add<T>(_ operation: PromiseOperation<T>) -> Promise<T> {
        return Promise(value: ()).then(on: queue) { Void -> Promise<T> in
            return operation.runAndWait()
        }
    }

    /// Submits a barrier operation for asynchronous execution.
    ///
    /// Submits an operation that will wait for all previously scheduled operations
    /// in the queue to finish before starting, and prevents newer operations from
    /// starting until the operation finishes.
    ///
    /// - Parameter operation: The operation to run as a barrier.
    ///
    /// - Returns: The `Promise` returned by running `operation`.
    public func addBarrier<T>(_ operation: PromiseOperation<T>) -> Promise<T> {
        return Promise(value: ()).then(on: .global()) { Void -> Promise<T> in
            var promise: Promise<T>!
            self.queue.sync(flags: .barrier, execute: {
                promise = operation.runAndWait()
            })
            return promise!
        }
    }

    /// Submits a barrier block for asynchronous execution.
    ///
    /// Submits an operation that will wait for all previously scheduled operations
    /// in the queue to finish before starting, and prevents newer operations from
    /// starting until the `Promise` produced by `makePromise` resolves.
    ///
    /// - Parameter makePromise: The block to run as a barrier.
    ///
    /// - Returns: The `Promise` returned by running the block.
    public func addBarrier<T>(_ makePromise: @escaping () -> Promise<T>) -> Promise<T> {
        return addBarrier(PromiseOperation(makePromise: makePromise))
    }
}