From 99faafe7234c6fa9780b91c8d4b5a99263723cce Mon Sep 17 00:00:00 2001
From: Alexander Levin <alevin.ru@gmail.com>
Date: Mon, 6 Mar 2017 19:57:19 +0200
Subject: [PATCH] SocketAckManager thread-safe acks removal

---
 Source/SocketAckManager.swift | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/Source/SocketAckManager.swift b/Source/SocketAckManager.swift
index ae6a6fa..eea183b 100644
--- a/Source/SocketAckManager.swift
+++ b/Source/SocketAckManager.swift
@@ -51,6 +51,7 @@ private func ==(lhs: SocketAck, rhs: SocketAck) -> Bool {
 
 struct SocketAckManager {
     private var acks = Set<SocketAck>(minimumCapacity: 1)
+    private let ackSemaphore = DispatchSemaphore(value: 1)
     
     mutating func addAck(_ ack: Int, callback: @escaping AckCallback) {
         acks.insert(SocketAck(ack: ack, callback: callback))
@@ -58,6 +59,8 @@ struct SocketAckManager {
     
     /// Should be called on handle queue
     mutating func executeAck(_ ack: Int, with items: [Any], onQueue: DispatchQueue) {
+        ackSemaphore.wait()
+        defer { ackSemaphore.signal() }
         let ack = acks.remove(SocketAck(ack: ack))
         
         onQueue.async() { ack?.callback(items) }
@@ -65,8 +68,12 @@ struct SocketAckManager {
     
     /// Should be called on handle queue
     mutating func timeoutAck(_ ack: Int, onQueue: DispatchQueue) {
+        ackSemaphore.wait()
+        defer { ackSemaphore.signal() }
         let ack = acks.remove(SocketAck(ack: ack))
         
-        onQueue.async() { ack?.callback?(["NO ACK"]) }
+        onQueue.async() {
+            ack?.callback?(["NO ACK"])
+        }
     }
 }