Merge branch 'refactor-engine' into development

* refactor-engine:
  remove refactor build
  Open client
  Make client single queued
  Make engine single queued
  add ignore for appcode

.gitignore

@@ -46,3 +46,5 @@ DerivedData
 *.xcuserstate
 
 Socket.IO-Test-Server/node_modules/*
+
+.idea/

Source/SocketAckEmitter.swift

@@ -22,6 +22,7 @@
 //  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 //  THE SOFTWARE.
 
+import Dispatch
 import Foundation
 
 public final class SocketAckEmitter : NSObject {
@@ -31,57 +32,52 @@ public final class SocketAckEmitter : NSObject {
     public var expected: Bool {
         return ackNum != -1
     }
-    
+
     init(socket: SocketIOClient, ackNum: Int) {
         self.socket = socket
         self.ackNum = ackNum
     }
-    
+
     public func with(_ items: SocketData...) {
         guard ackNum != -1 else { return }
-        
+
         socket.emitAck(ackNum, with: items)
     }
-    
+
     public func with(_ items: [Any]) {
         guard ackNum != -1 else { return }
-        
+
         socket.emitAck(ackNum, with: items)
     }
-        
+
 }
 
 public final class OnAckCallback : NSObject {
     private let ackNumber: Int
     private let items: [Any]
     private weak var socket: SocketIOClient?
-    
+
     init(ackNumber: Int, items: [Any], socket: SocketIOClient) {
         self.ackNumber = ackNumber
         self.items = items
         self.socket = socket
     }
-    
+
     deinit {
         DefaultSocketLogger.Logger.log("OnAckCallback for \(ackNumber) being released", type: "OnAckCallback")
     }
-    
+
     public func timingOut(after seconds: Int, callback: @escaping AckCallback) {
         guard let socket = self.socket else { return }
-        
+
-        socket.ackQueue.sync() {
+        socket.ackHandlers.addAck(ackNumber, callback: callback)
-            socket.ackHandlers.addAck(ackNumber, callback: callback)
-        }
-        
         socket._emit(items, ack: ackNumber)
-        
+
         guard seconds != 0 else { return }
-        
+
-        let time = DispatchTime.now() + Double(UInt64(seconds) * NSEC_PER_SEC) / Double(NSEC_PER_SEC)
+        socket.handleQueue.asyncAfter(deadline: DispatchTime.now() + Double(seconds)) {
-        
-        socket.handleQueue.asyncAfter(deadline: time) {
             socket.ackHandlers.timeoutAck(self.ackNumber, onQueue: socket.handleQueue)
         }
     }
-    
+
 }

Source/SocketEngine.swift

@@ -22,12 +22,11 @@
 //  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 //  THE SOFTWARE.
 
+import Dispatch
 import Foundation
 
 public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePollable, SocketEngineWebsocket {
-    public let emitQueue = DispatchQueue(label: "com.socketio.engineEmitQueue", attributes: [])
+    public let engineQueue = DispatchQueue(label: "com.socketio.engineHandleQueue", attributes: [])
-    public let handleQueue = DispatchQueue(label: "com.socketio.engineHandleQueue", attributes: [])
-    public let parseQueue = DispatchQueue(label: "com.socketio.engineParseQueue", attributes: [])
 
     public var connectParams: [String: Any]? {
         didSet {
@@ -101,7 +100,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
                 forceWebsockets = force
             case let .path(path):
                 socketPath = path
-                
+
                 if !socketPath.hasSuffix("/") {
                     socketPath += "/"
                 }
@@ -119,12 +118,12 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
         }
 
         super.init()
-        
+
         sessionDelegate = sessionDelegate ?? self
-        
+
         (urlPolling, urlWebSocket) = createURLs()
     }
-    
+
     public convenience init(client: SocketEngineClient, url: URL, options: NSDictionary?) {
         self.init(client: client, url: url, config: options?.toSocketConfiguration() ?? [])
     }
@@ -139,7 +138,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
         do {
             let dict = try msg.toNSDictionary()
             guard let error = dict["message"] as? String else { return }
-            
+
             /*
              0: Unknown transport
              1: Unknown sid
@@ -155,7 +154,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     private func handleBase64(message: String) {
         // binary in base64 string
         let noPrefix = message[message.index(message.startIndex, offsetBy: 2)..<message.endIndex]
-        
+
         if let data = NSData(base64Encoded: noPrefix, options: .ignoreUnknownCharacters) {
             client?.parseEngineBinaryData(data as Data)
         }
@@ -174,6 +173,12 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 
     /// Starts the connection to the server
     public func connect() {
+        engineQueue.async {
+            self._connect()
+        }
+    }
+
+    private func _connect() {
         if connected {
             DefaultSocketLogger.Logger.error("Engine tried opening while connected. Assuming this was a reconnect", type: logType)
             disconnect(reason: "reconnect")
@@ -191,8 +196,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
             return
         }
 
-        var reqPolling = URLRequest(url: urlPolling, cachePolicy: .reloadIgnoringLocalCacheData,
+        var reqPolling = URLRequest(url: urlPolling, cachePolicy: .reloadIgnoringLocalCacheData, timeoutInterval: 60.0)
-                                    timeoutInterval: 60.0)
 
         if cookies != nil {
             let headers = HTTPCookie.requestHeaderFields(with: cookies!)
@@ -216,7 +220,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
         var urlPolling = URLComponents(string: url.absoluteString)!
         var urlWebSocket = URLComponents(string: url.absoluteString)!
         var queryString = ""
-        
+
         urlWebSocket.path = socketPath
         urlPolling.path = socketPath
 
@@ -244,7 +248,6 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     }
 
     private func createWebsocketAndConnect() {
-        
         ws?.delegate = nil
         ws = WebSocket(url: urlWebSocketWithSid as URL)
 
@@ -261,7 +264,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
             }
         }
 
-        ws?.callbackQueue = handleQueue
+        ws?.callbackQueue = engineQueue
         ws?.voipEnabled = voipEnabled
         ws?.delegate = self
         ws?.disableSSLCertValidation = selfSigned
@@ -277,6 +280,12 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     }
 
     public func disconnect(reason: String) {
+        engineQueue.async {
+            self._disconnect(reason: reason)
+        }
+    }
+
+    private func _disconnect(reason: String) {
         guard connected else { return closeOutEngine(reason: reason) }
 
         DefaultSocketLogger.Logger.log("Engine is being closed.", type: logType)
@@ -296,12 +305,10 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     // We need to take special care when we're polling that we send it ASAP
     // Also make sure we're on the emitQueue since we're touching postWait
     private func disconnectPolling(reason: String) {
-        emitQueue.sync {
+        postWait.append(String(SocketEnginePacketType.close.rawValue))
-            self.postWait.append(String(SocketEnginePacketType.close.rawValue))
+
-            let req = self.createRequestForPostWithPostWait()
+        doRequest(for: createRequestForPostWithPostWait()) {_, _, _ in }
-            self.doRequest(for: req) {_, _, _ in }
+        closeOutEngine(reason: reason)
-            self.closeOutEngine(reason: reason)
-        }
     }
 
     public func doFastUpgrade() {
@@ -321,16 +328,14 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     private func flushProbeWait() {
         DefaultSocketLogger.Logger.log("Flushing probe wait", type: logType)
 
-        emitQueue.async {
+        for waiter in probeWait {
-            for waiter in self.probeWait {
+            write(waiter.msg, withType: waiter.type, withData: waiter.data)
-                self.write(waiter.msg, withType: waiter.type, withData: waiter.data)
+        }
-            }
 
-            self.probeWait.removeAll(keepingCapacity: false)
+        probeWait.removeAll(keepingCapacity: false)
 
-            if self.postWait.count != 0 {
+        if postWait.count != 0 {
-                self.flushWaitingForPostToWebSocket()
+            flushWaitingForPostToWebSocket()
-            }
         }
     }
 
@@ -357,47 +362,47 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
     private func handleNOOP() {
         doPoll()
     }
-    
+
     private func handleOpen(openData: String) {
         guard let json = try? openData.toNSDictionary() else {
             didError(reason: "Error parsing open packet")
-            
+
             return
         }
-        
+
         guard let sid = json["sid"] as? String else {
             didError(reason: "Open packet contained no sid")
-            
+
             return
         }
-        
+
         let upgradeWs: Bool
-        
+
         self.sid = sid
         connected = true
         pongsMissed = 0
-        
+
         if let upgrades = json["upgrades"] as? [String] {
             upgradeWs = upgrades.contains("websocket")
         } else {
             upgradeWs = false
         }
-        
+
         if let pingInterval = json["pingInterval"] as? Double, let pingTimeout = json["pingTimeout"] as? Double {
             self.pingInterval = pingInterval / 1000.0
             self.pingTimeout = pingTimeout / 1000.0
         }
-        
+
         if !forcePolling && !forceWebsockets && upgradeWs {
             createWebsocketAndConnect()
         }
-        
+
         sendPing()
-        
+
         if !forceWebsockets {
             doPoll()
         }
-        
+
         client?.engineDidOpen(reason: "Connect")
     }
 
@@ -421,14 +426,14 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 
         let reader = SocketStringReader(message: message)
         let fixedString: String
-        
+
         if message.hasPrefix("b4") {
             return handleBase64(message: message)
         }
 
         guard let type = SocketEnginePacketType(rawValue: Int(reader.currentCharacter) ?? -1) else {
             checkAndHandleEngineError(message)
-            
+
             return
         }
 
@@ -456,13 +461,16 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 
     // Puts the engine back in its default state
     private func resetEngine() {
+        let queue = OperationQueue()
+        queue.underlyingQueue = engineQueue
+
         closed = false
         connected = false
         fastUpgrade = false
         polling = true
         probing = false
         invalidated = false
-        session = Foundation.URLSession(configuration: .default, delegate: sessionDelegate, delegateQueue: OperationQueue.main)
+        session = Foundation.URLSession(configuration: .default, delegate: sessionDelegate, delegateQueue: queue)
         sid = ""
         waitingForPoll = false
         waitingForPost = false
@@ -475,17 +483,16 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
         // Server is not responding
         if pongsMissed > pongsMissedMax {
             client?.engineDidClose(reason: "Ping timeout")
-            
+
             return
         }
 
         guard let pingInterval = pingInterval else { return }
-        
+
         pongsMissed += 1
         write("", withType: .ping, withData: [])
-        
+
-        let time = DispatchTime.now() + Double(Int64(pingInterval * Double(NSEC_PER_SEC))) / Double(NSEC_PER_SEC)
+        engineQueue.asyncAfter(deadline: DispatchTime.now() + Double(pingInterval)) {[weak self] in self?.sendPing() }
-        DispatchQueue.main.asyncAfter(deadline: time) {[weak self] in self?.sendPing() }
     }
 
     // Moves from long-polling to websockets
@@ -501,16 +508,16 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 
     /// Write a message, independent of transport.
     public func write(_ msg: String, withType type: SocketEnginePacketType, withData data: [Data]) {
-        emitQueue.async {
+        engineQueue.async {
             guard self.connected else { return }
 
             if self.websocket {
                 DefaultSocketLogger.Logger.log("Writing ws: %@ has data: %@",
-                    type: self.logType, args: msg, data.count != 0)
+                                               type: self.logType, args: msg, data.count != 0)
                 self.sendWebSocketMessage(msg, withType: type, withData: data)
             } else if !self.probing {
                 DefaultSocketLogger.Logger.log("Writing poll: %@ has data: %@",
-                    type: self.logType, args: msg, data.count != 0)
+                                               type: self.logType, args: msg, data.count != 0)
                 self.sendPollMessage(msg, withType: type, withData: data)
             } else {
                 self.probeWait.append((msg, type, data))
@@ -535,7 +542,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 
         if closed {
             client?.engineDidClose(reason: "Disconnect")
-            
+
             return
         }
 
@@ -557,7 +564,7 @@ public final class SocketEngine : NSObject, URLSessionDelegate, SocketEnginePoll
 extension SocketEngine {
     public func URLSession(session: URLSession, didBecomeInvalidWithError error: NSError?) {
         DefaultSocketLogger.Logger.error("Engine URLSession became invalid", type: "SocketEngine")
-        
+
         didError(reason: "Engine URLSession became invalid")
     }
 }

Source/SocketEnginePollable.swift

@@ -27,7 +27,7 @@ import Foundation
 /// Protocol that is used to implement socket.io polling support
 public protocol SocketEnginePollable : SocketEngineSpec {
     var invalidated: Bool { get }
-    /// Holds strings waiting to be sent over polling. 
+    /// Holds strings waiting to be sent over polling.
     /// You shouldn't need to mess with this.
     var postWait: [String] { get set }
     var session: URLSession? { get }
@@ -37,7 +37,7 @@ public protocol SocketEnginePollable : SocketEngineSpec {
     /// Because socket.io doesn't let you send two post request at the same time
     /// we have to keep track if there's an outstanding post
     var waitingForPost: Bool { get set }
-    
+
     func doPoll()
     func sendPollMessage(_ message: String, withType type: SocketEnginePacketType, withData datas: [Data])
     func stopPolling()
@@ -47,95 +47,94 @@ public protocol SocketEnginePollable : SocketEngineSpec {
 extension SocketEnginePollable {
     private func addHeaders(for req: URLRequest) -> URLRequest {
         var req = req
-        
+
         if cookies != nil {
             let headers = HTTPCookie.requestHeaderFields(with: cookies!)
             req.allHTTPHeaderFields = headers
         }
-        
+
         if extraHeaders != nil {
             for (headerName, value) in extraHeaders! {
                 req.setValue(value, forHTTPHeaderField: headerName)
             }
         }
-        
+
         return req
     }
-    
+
     func createRequestForPostWithPostWait() -> URLRequest {
         defer { postWait.removeAll(keepingCapacity: true) }
 
         var postStr = ""
-        
+
         for packet in postWait {
             let len = packet.characters.count
-            
+
             postStr += "\(len):\(packet)"
         }
-        
+
         DefaultSocketLogger.Logger.log("Created POST string: %@", type: "SocketEnginePolling", args: postStr)
-        
+
         var req = URLRequest(url: urlPollingWithSid)
         let postData = postStr.data(using: .utf8, allowLossyConversion: false)!
-        
+
         req = addHeaders(for: req)
-        
+
         req.httpMethod = "POST"
         req.setValue("text/plain; charset=UTF-8", forHTTPHeaderField: "Content-Type")
 
         req.httpBody = postData
         req.setValue(String(postData.count), forHTTPHeaderField: "Content-Length")
-        
+
-        return req as URLRequest
+        return req
     }
-    
+
     public func doPoll() {
         if websocket || waitingForPoll || !connected || closed {
             return
         }
-        
+
-        waitingForPoll = true
-        
         var req = URLRequest(url: urlPollingWithSid)
-        
         req = addHeaders(for: req)
+
         doLongPoll(for: req )
     }
-    
+
     func doRequest(for req: URLRequest, callbackWith callback: @escaping (Data?, URLResponse?, Error?) -> Void) {
         if !polling || closed || invalidated || fastUpgrade {
             return
         }
-        
+
-        DefaultSocketLogger.Logger.log("Doing polling request", type: "SocketEnginePolling")
+        DefaultSocketLogger.Logger.log("Doing polling %@ %@", type: "SocketEnginePolling",
-        
+                                       args: req.httpMethod ?? "", req)
+
         session?.dataTask(with: req, completionHandler: callback).resume()
     }
-    
+
     func doLongPoll(for req: URLRequest) {
+        waitingForPoll = true
+
         doRequest(for: req) {[weak self] data, res, err in
             guard let this = self, this.polling else { return }
-            
+
             if err != nil || data == nil {
                 DefaultSocketLogger.Logger.error(err?.localizedDescription ?? "Error", type: "SocketEnginePolling")
-                
+
                 if this.polling {
                     this.didError(reason: err?.localizedDescription ?? "Error")
                 }
-                
+
                 return
             }
-            
+
             DefaultSocketLogger.Logger.log("Got polling response", type: "SocketEnginePolling")
-            
+
             if let str = String(data: data!, encoding: String.Encoding.utf8) {
-                this.parseQueue.async {
+                this.parsePollingMessage(str)
-                    this.parsePollingMessage(str)
-                }
             }
-            
+
             this.waitingForPoll = false
-            
+
             if this.fastUpgrade {
                 this.doFastUpgrade()
             } else if !this.closed && this.polling {
@@ -143,7 +142,7 @@ extension SocketEnginePollable {
             }
         }
     }
-    
+
     private func flushWaitingForPost() {
         if postWait.count == 0 || !connected {
             return
@@ -151,79 +150,75 @@ extension SocketEnginePollable {
             flushWaitingForPostToWebSocket()
             return
         }
-        
+
         let req = createRequestForPostWithPostWait()
-        
+
         waitingForPost = true
-        
+
         DefaultSocketLogger.Logger.log("POSTing", type: "SocketEnginePolling")
-        
+
         doRequest(for: req) {[weak self] data, res, err in
             guard let this = self else { return }
-            
+
             if err != nil {
                 DefaultSocketLogger.Logger.error(err?.localizedDescription ?? "Error", type: "SocketEnginePolling")
-                
+
                 if this.polling {
                     this.didError(reason: err?.localizedDescription ?? "Error")
                 }
-                
+
                 return
             }
-            
+
             this.waitingForPost = false
-            
+
-            this.emitQueue.async {
+            if !this.fastUpgrade {
-                if !this.fastUpgrade {
+                this.flushWaitingForPost()
-                    this.flushWaitingForPost()
+                this.doPoll()
-                    this.doPoll()
-                }
             }
         }
     }
-    
+
     func parsePollingMessage(_ str: String) {
         guard str.characters.count != 1 else { return }
-        
+
         var reader = SocketStringReader(message: str)
-        
+
         while reader.hasNext {
             if let n = Int(reader.readUntilOccurence(of: ":")) {
-                let str = reader.read(count: n)
+                parseEngineMessage(reader.read(count: n), fromPolling: true)
-                
-                handleQueue.async { self.parseEngineMessage(str, fromPolling: true) }
             } else {
-                handleQueue.async { self.parseEngineMessage(str, fromPolling: true) }
+                parseEngineMessage(str, fromPolling: true)
                 break
             }
         }
     }
-    
+
     /// Send polling message.
     /// Only call on emitQueue
     public func sendPollMessage(_ message: String, withType type: SocketEnginePacketType, withData datas: [Data]) {
         DefaultSocketLogger.Logger.log("Sending poll: %@ as type: %@", type: "SocketEnginePolling", args: message, type.rawValue)
         let fixedMessage: String
-        
+
         if doubleEncodeUTF8 {
             fixedMessage = doubleEncodeUTF8(message)
         } else {
             fixedMessage = message
         }
-        
+
         postWait.append(String(type.rawValue) + fixedMessage)
-        
+
         for data in datas {
             if case let .right(bin) = createBinaryDataForSend(using: data) {
                 postWait.append(bin)
             }
         }
-        
+
         if !waitingForPost {
             flushWaitingForPost()
         }
     }
-    
+
     public func stopPolling() {
         waitingForPoll = false
         waitingForPost = false

Source/SocketEngineSpec.swift

@@ -32,15 +32,13 @@ import Foundation
     var connectParams: [String: Any]? { get set }
     var doubleEncodeUTF8: Bool { get }
     var cookies: [HTTPCookie]? { get }
+    var engineQueue: DispatchQueue { get }
     var extraHeaders: [String: String]? { get }
     var fastUpgrade: Bool { get }
     var forcePolling: Bool { get }
     var forceWebsockets: Bool { get }
-    var parseQueue: DispatchQueue { get }
     var polling: Bool { get }
     var probing: Bool { get }
-    var emitQueue: DispatchQueue { get }
-    var handleQueue: DispatchQueue { get }
     var sid: String { get }
     var socketPath: String { get }
     var urlPolling: URL { get }

Source/SocketIOClient.swift

@@ -22,9 +22,10 @@
 //  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 //  THE SOFTWARE.
 
+import Dispatch
 import Foundation
 
-public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineClient, SocketParsable {
+open class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineClient, SocketParsable {
     public let socketURL: URL
 
     public private(set) var engine: SocketEngineSpec?
@@ -41,43 +42,38 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     public var forceNew = false
+    public var handleQueue = DispatchQueue.main
     public var nsp = "/"
     public var config: SocketIOClientConfiguration
     public var reconnects = true
     public var reconnectWait = 10
 
     private let logType = "SocketIOClient"
-    private let parseQueue = DispatchQueue(label: "com.socketio.parseQueue")
 
     private var anyHandler: ((SocketAnyEvent) -> Void)?
     private var currentReconnectAttempt = 0
     private var handlers = [SocketEventHandler]()
     private var reconnecting = false
 
-    private let ackSemaphore = DispatchSemaphore(value: 1)
     private(set) var currentAck = -1
-    private(set) var handleQueue = DispatchQueue.main
     private(set) var reconnectAttempts = -1
-    
-    let ackQueue = DispatchQueue(label: "com.socketio.ackQueue")
-    let emitQueue = DispatchQueue(label: "com.socketio.emitQueue")
 
     var ackHandlers = SocketAckManager()
     var waitingPackets = [SocketPacket]()
-    
+
     public var sid: String? {
         return engine?.sid
     }
-    
+
     /// Type safe way to create a new SocketIOClient. opts can be omitted
     public init(socketURL: URL, config: SocketIOClientConfiguration = []) {
         self.config = config
         self.socketURL = socketURL
-        
+
         if socketURL.absoluteString.hasPrefix("https://") {
             self.config.insert(.secure(true))
         }
-        
+
         for option in config {
             switch option {
             case let .reconnects(reconnects):
@@ -102,10 +98,10 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
         }
 
         self.config.insert(.path("/socket.io/"), replacing: false)
-        
+
         super.init()
     }
-    
+
     /// Not so type safe way to create a SocketIOClient, meant for Objective-C compatiblity.
     /// If using Swift it's recommended to use `init(socketURL: NSURL, options: Set<SocketIOClientOption>)`
     public convenience init(socketURL: NSURL, config: NSDictionary?) {
@@ -127,13 +123,13 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     /// Connect to the server.
-    public func connect() {
+    open func connect() {
         connect(timeoutAfter: 0, withHandler: nil)
     }
 
     /// Connect to the server. If we aren't connected after timeoutAfter, call withHandler
     /// 0 Never times out
-    public func connect(timeoutAfter: Int, withHandler handler: (() -> Void)?) {
+    open func connect(timeoutAfter: Int, withHandler handler: (() -> Void)?) {
         assert(timeoutAfter >= 0, "Invalid timeout: \(timeoutAfter)")
 
         guard status != .connected else {
@@ -148,30 +144,25 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
         } else {
             engine?.connect()
         }
-        
+
         guard timeoutAfter != 0 else { return }
 
         let time = DispatchTime.now() + Double(UInt64(timeoutAfter) * NSEC_PER_SEC) / Double(NSEC_PER_SEC)
 
         handleQueue.asyncAfter(deadline: time) {[weak self] in
             guard let this = self, this.status != .connected && this.status != .disconnected else { return }
-            
+
             this.status = .disconnected
             this.engine?.disconnect(reason: "Connect timeout")
-            
+
             handler?()
         }
     }
 
-    private func nextAck() -> Int {
-        ackSemaphore.wait()
-        defer { ackSemaphore.signal() }
-        currentAck += 1
-        return currentAck
-    }
-
     private func createOnAck(_ items: [Any]) -> OnAckCallback {
-        return OnAckCallback(ackNumber: nextAck(), items: items, socket: self)
+        currentAck += 1
+
+        return OnAckCallback(ackNumber: currentAck, items: items, socket: self)
     }
 
     func didConnect() {
@@ -197,73 +188,73 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     /// Disconnects the socket.
-    public func disconnect() {
+    open func disconnect() {
         DefaultSocketLogger.Logger.log("Closing socket", type: logType)
 
         didDisconnect(reason: "Disconnect")
     }
 
     /// Send a message to the server
-    public func emit(_ event: String, _ items: SocketData...) {
+    open func emit(_ event: String, _ items: SocketData...) {
         emit(event, with: items)
     }
 
     /// Same as emit, but meant for Objective-C
-    public func emit(_ event: String, with items: [Any]) {
+    open func emit(_ event: String, with items: [Any]) {
         guard status == .connected else {
             handleEvent("error", data: ["Tried emitting \(event) when not connected"], isInternalMessage: true)
             return
         }
-        
+
         _emit([event] + items)
     }
 
     /// Sends a message to the server, requesting an ack. Use the onAck method of SocketAckHandler to add
     /// an ack.
-    public func emitWithAck(_ event: String, _ items: SocketData...) -> OnAckCallback {
+    open func emitWithAck(_ event: String, _ items: SocketData...) -> OnAckCallback {
         return emitWithAck(event, with: items)
     }
 
     /// Same as emitWithAck, but for Objective-C
-    public func emitWithAck(_ event: String, with items: [Any]) -> OnAckCallback {
+    open func emitWithAck(_ event: String, with items: [Any]) -> OnAckCallback {
         return createOnAck([event] + items)
     }
 
     func _emit(_ data: [Any], ack: Int? = nil) {
-        emitQueue.async {
+        guard status == .connected else {
-            guard self.status == .connected else {
+            handleEvent("error", data: ["Tried emitting when not connected"], isInternalMessage: true)
-                self.handleEvent("error", data: ["Tried emitting when not connected"], isInternalMessage: true)
+            return
-                return
-            }
-            
-            let packet = SocketPacket.packetFromEmit(data, id: ack ?? -1, nsp: self.nsp, ack: false)
-            let str = packet.packetString
-            
-            DefaultSocketLogger.Logger.log("Emitting: %@", type: self.logType, args: str)
-            
-            self.engine?.send(str, withData: packet.binary)
         }
+
+        let packet = SocketPacket.packetFromEmit(data, id: ack ?? -1, nsp: nsp, ack: false)
+        let str = packet.packetString
+
+        DefaultSocketLogger.Logger.log("Emitting: %@", type: logType, args: str)
+
+        engine?.send(str, withData: packet.binary)
     }
 
     // If the server wants to know that the client received data
     func emitAck(_ ack: Int, with items: [Any]) {
-        emitQueue.async {
+        guard status == .connected else { return }
-            guard self.status == .connected else { return }
+
-            
+        let packet = SocketPacket.packetFromEmit(items, id: ack, nsp: nsp, ack: true)
-            let packet = SocketPacket.packetFromEmit(items, id: ack, nsp: self.nsp, ack: true)
+        let str = packet.packetString
-            let str = packet.packetString
+
-            
+        DefaultSocketLogger.Logger.log("Emitting Ack: %@", type: logType, args: str)
-            DefaultSocketLogger.Logger.log("Emitting Ack: %@", type: self.logType, args: str)
+
-            
+        engine?.send(str, withData: packet.binary)
-            self.engine?.send(str, withData: packet.binary)
+    }
+
+    open func engineDidClose(reason: String) {
+        handleQueue.async {
+            self._engineDidClose(reason: reason)
         }
     }
 
-    public func engineDidClose(reason: String) {
+    private func _engineDidClose(reason: String) {
-        parseQueue.async {
+        waitingPackets.removeAll()
-            self.waitingPackets.removeAll()
+
-        }
-        
         if status != .disconnected {
             status = .notConnected
         }
@@ -276,14 +267,20 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
         }
     }
 
+    open func engineDidError(reason: String) {
+        handleQueue.async {
+            self._engineDidError(reason: reason)
+        }
+    }
+
     /// error
-    public func engineDidError(reason: String) {
+    private func _engineDidError(reason: String) {
         DefaultSocketLogger.Logger.error("%@", type: logType, args: reason)
 
         handleEvent("error", data: [reason], isInternalMessage: true)
     }
-    
+
-    public func engineDidOpen(reason: String) {
+    open func engineDidOpen(reason: String) {
         DefaultSocketLogger.Logger.log(reason, type: "SocketEngineClient")
     }
 
@@ -293,28 +290,24 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
 
         DefaultSocketLogger.Logger.log("Handling ack: %@ with data: %@", type: logType, args: ack, data)
 
-        handleQueue.async() {
+        ackHandlers.executeAck(ack, with: data, onQueue: handleQueue)
-            self.ackHandlers.executeAck(ack, with: data, onQueue: self.handleQueue)
-        }
     }
 
     /// Causes an event to be handled. Only use if you know what you're doing.
-    public func handleEvent(_ event: String, data: [Any], isInternalMessage: Bool, withAck ack: Int = -1) {
+    open func handleEvent(_ event: String, data: [Any], isInternalMessage: Bool, withAck ack: Int = -1) {
         guard status == .connected || isInternalMessage else { return }
 
         DefaultSocketLogger.Logger.log("Handling event: %@ with data: %@", type: logType, args: event, data)
 
-        handleQueue.async {
+        anyHandler?(SocketAnyEvent(event: event, items: data))
-            self.anyHandler?(SocketAnyEvent(event: event, items: data))
 
-            for handler in self.handlers where handler.event == event {
+        for handler in handlers where handler.event == event {
-                handler.executeCallback(with: data, withAck: ack, withSocket: self)
+            handler.executeCallback(with: data, withAck: ack, withSocket: self)
-            }
         }
     }
 
     /// Leaves nsp and goes back to /
-    public func leaveNamespace() {
+    open func leaveNamespace() {
         if nsp != "/" {
             engine?.send("1\(nsp)", withData: [])
             nsp = "/"
@@ -322,7 +315,7 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     /// Joins namespace
-    public func joinNamespace(_ namespace: String) {
+    open func joinNamespace(_ namespace: String) {
         nsp = namespace
 
         if nsp != "/" {
@@ -332,14 +325,14 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     /// Removes handler(s) based on name
-    public func off(_ event: String) {
+    open func off(_ event: String) {
         DefaultSocketLogger.Logger.log("Removing handler for event: %@", type: logType, args: event)
 
         handlers = handlers.filter({ $0.event != event })
     }
 
     /// Removes a handler with the specified UUID gotten from an `on` or `once`
-    public func off(id: UUID) {
+    open func off(id: UUID) {
         DefaultSocketLogger.Logger.log("Removing handler with id: %@", type: logType, args: id)
 
         handlers = handlers.filter({ $0.id != id })
@@ -348,7 +341,7 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     /// Adds a handler for an event.
     /// Returns: A unique id for the handler
     @discardableResult
-    public func on(_ event: String, callback: @escaping NormalCallback) -> UUID {
+    open func on(_ event: String, callback: @escaping NormalCallback) -> UUID {
         DefaultSocketLogger.Logger.log("Adding handler for event: %@", type: logType, args: event)
 
         let handler = SocketEventHandler(event: event, id: UUID(), callback: callback)
@@ -360,7 +353,7 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     /// Adds a single-use handler for an event.
     /// Returns: A unique id for the handler
     @discardableResult
-    public func once(_ event: String, callback: @escaping NormalCallback) -> UUID {
+    open func once(_ event: String, callback: @escaping NormalCallback) -> UUID {
         DefaultSocketLogger.Logger.log("Adding once handler for event: %@", type: logType, args: event)
 
         let id = UUID()
@@ -377,30 +370,30 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
     }
 
     /// Adds a handler that will be called on every event.
-    public func onAny(_ handler: @escaping (SocketAnyEvent) -> Void) {
+    open func onAny(_ handler: @escaping (SocketAnyEvent) -> Void) {
         anyHandler = handler
     }
 
     public func parseEngineMessage(_ msg: String) {
         DefaultSocketLogger.Logger.log("Should parse message: %@", type: "SocketIOClient", args: msg)
 
-        parseQueue.async { self.parseSocketMessage(msg) }
+        handleQueue.async { self.parseSocketMessage(msg) }
     }
 
     public func parseEngineBinaryData(_ data: Data) {
-        parseQueue.async { self.parseBinaryData(data) }
+        handleQueue.async { self.parseBinaryData(data) }
     }
 
     /// Tries to reconnect to the server.
-    public func reconnect() {
+    open func reconnect() {
         guard !reconnecting else { return }
-        
+
         engine?.disconnect(reason: "manual reconnect")
     }
 
     /// Removes all handlers.
     /// Can be used after disconnecting to break any potential remaining retain cycles.
-    public func removeAllHandlers() {
+    open func removeAllHandlers() {
         handlers.removeAll(keepingCapacity: false)
     }
 
@@ -409,7 +402,7 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
 
         DefaultSocketLogger.Logger.log("Starting reconnect", type: logType)
         handleEvent("reconnect", data: [reason], isInternalMessage: true)
-        
+
         _tryReconnect()
     }
 
@@ -425,26 +418,24 @@ public final class SocketIOClient : NSObject, SocketIOClientSpec, SocketEngineCl
 
         currentReconnectAttempt += 1
         connect()
-        
+
-        let deadline = DispatchTime.now() + Double(Int64(UInt64(reconnectWait) * NSEC_PER_SEC)) / Double(NSEC_PER_SEC)
+        handleQueue.asyncAfter(deadline: DispatchTime.now() + Double(reconnectWait), execute: _tryReconnect)
-        
-        DispatchQueue.main.asyncAfter(deadline: deadline, execute: _tryReconnect)
     }
-    
+
     // Test properties
-    
+
     var testHandlers: [SocketEventHandler] {
         return handlers
     }
-    
+
     func setTestable() {
         status = .connected
     }
-    
+
     func setTestEngine(_ engine: SocketEngineSpec?) {
         self.engine = engine
     }
-    
+
     func emitTest(event: String, _ data: Any...) {
         _emit([event] + data)
     }

Source/SocketIOClientSpec.swift

@@ -22,10 +22,13 @@
 //  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 //  THE SOFTWARE.
 
+import Dispatch
+
 protocol SocketIOClientSpec : class {
+    var handleQueue: DispatchQueue { get set }
     var nsp: String { get set }
     var waitingPackets: [SocketPacket] { get set }
-    
+
     func didConnect()
     func didDisconnect(reason: String)
     func didError(reason: String)
@@ -37,7 +40,7 @@ protocol SocketIOClientSpec : class {
 extension SocketIOClientSpec {
     func didError(reason: String) {
         DefaultSocketLogger.Logger.error("%@", type: "SocketIOClient", args: reason)
-        
+
         handleEvent("error", data: [reason], isInternalMessage: true, withAck: -1)
     }
 }