Calling functions in the serial queue in Swift

I've got a function which is called by observing the NotificationCenter:

NotificationCenter.default.addObserver(self, selector: #selector(observedPosition(_: ), name: "calculatePosition", object: nil)

and then the function:

@objc func observedPosition(_ notification: NSNotification) {
   if let data = notification.object as? Int {
      self.sendPosition(from: data)
   }

As this function can be called multiple times in very short time periods I would like to add it to the queue and call sendPosition() only once the previous sendPosition() has finished.

I tried something like this but dunno if it's a correct approach:

@objc func observedPosition(_ notification: NSNotification) {
    let queue = DispatchQueue(label: queueLabel, attributes: [], targer: nil)
    queue.sync {
        if let data = notification.object as? Int {
            self.sendPosition(from: data)
        }
    }
}

Upvotes: 3

Answers (2)

Vasily Bodnarchuk

Reputation: 25294

Details

Xcode Version 10.3 (10G8), Swift 5

Key features

Implemented own queue which will execute functions one by one
All operations (closures) stored in array
Thread safety

Solution

// MARK: - StackableOperationsQueue performs functions from the stack one by one (serial performing)

class StackableOperationsQueue {
    private let semaphore = DispatchSemaphore(value: 1)
    private lazy var operations = [QueueOperation]()
    private lazy var isExecuting = false

    fileprivate func _append(operation: QueueOperation) {
        semaphore.wait()
        operations.append(operation)
        semaphore.signal()
        execute()
    }

    func append(operation: QueueOperation) { _append(operation: operation) }

    private func execute() {
        semaphore.wait()
        guard !operations.isEmpty, !isExecuting else { semaphore.signal(); return }
        let operation = operations.removeFirst()
        isExecuting = true
        semaphore.signal()
        operation.perform()
        semaphore.wait()
        isExecuting = false
        semaphore.signal()
        execute()
    }
}

// MARK: - StackableOperationsCuncurentQueue performs functions from the stack one by one (serial performing) but in cuncurent queue

class StackableOperationsCuncurentQueue: StackableOperationsQueue {
    private var queue: DispatchQueue
    init(queue: DispatchQueue) { self.queue = queue }
    override func append(operation: QueueOperation) {
        queue.async { [weak self] in self?._append(operation: operation) }
    }
}

// MARK: QueueOperation interface

protocol QueueOperation: class {
    var сlosure: (() -> Void)? { get }
    var actualityCheckingClosure: (() -> Bool)? { get }
    init (actualityCheckingClosure: (() -> Bool)?, serialClosure: (() -> Void)?)
    func perform()
}

extension QueueOperation {
    // MARK: - Can queue perform the operation `сlosure: (() -> Void)?` or not
    var isActual: Bool {
        guard   let actualityCheckingClosure = self.actualityCheckingClosure,
                self.сlosure != nil else { return false }
        return actualityCheckingClosure()
    }
    func perform() { if isActual { сlosure?() } }

    init (actualIifNotNill object: AnyObject?, serialClosure: (() -> Void)?) {
        self.init(actualityCheckingClosure: { return object != nil }, serialClosure: serialClosure)
    }
}

class SerialQueueOperation: QueueOperation {
    let сlosure: (() -> Void)?
    let actualityCheckingClosure: (() -> Bool)?
    required init (actualityCheckingClosure: (() -> Bool)?, serialClosure: (() -> Void)?) {
        self.actualityCheckingClosure = actualityCheckingClosure
        self.сlosure = serialClosure
    }
}

Usage example

class TEST {

    private lazy var stackableOperationsQueue: StackableOperationsCuncurentQueue = {
        let queue = DispatchQueue(label: "custom_queue", qos: .background,
                                  attributes: [.concurrent], autoreleaseFrequency: .workItem, target: nil)
        return StackableOperationsCuncurentQueue(queue: queue)
    }()

    private func addOperationToQueue(closure: (() -> Void)?) {
        let operation = SerialQueueOperation(actualIifNotNill: self) { closure?() }
        stackableOperationsQueue.append(operation: operation)
        print("!!!! Function added ")
    }

    private func simpleFunc(index: Int) {
        print("Func \(index) started")
        sleep(UInt32(index+1));
        print("Func \(index) ended")
    }

    func run() {
        (0...3).forEach { index in
            addOperationToQueue { [weak self] in self?.simpleFunc(index: index) }
        }
    }
}

let test = TEST()
test.run()

Usage example results

//  qos: .background
!!!! Function added 
!!!! Function added 
!!!! Function added 
!!!! Function added 
Func 0 started
Func 0 ended
Func 1 started
Func 1 ended
Func 2 started
Func 2 ended
Func 3 started
Func 3 ended


//  qos: .userInitiated
!!!! Function added 
Func 0 started
!!!! Function added 
!!!! Function added 
!!!! Function added 
Func 0 ended
Func 1 started
Func 1 ended
Func 2 started
Func 2 ended
Func 3 started
Func 3 ended

Upvotes: 3

Alexander

Reputation: 63272

That is correct, so long as you ensure the same queue is being used to schedule all sendPosition method calls. For example, if this queue were a local variable, it would be of no use at all.

Upvotes: 1