redo acks

Author: nuclearace

README.md

@@ -11,7 +11,7 @@ socket.on("connect") {data, ack in
 
 socket.on("currentAmount") {data, ack in
     if let cur = data?[0] as? Double {
-        socket.emitWithAck("canUpdate", cur).onAck(0) {data in
+        socket.emitWithAck("canUpdate", cur)(timeout: 0) {data in
             socket.emit("update", ["amount": cur + 2.50])
         }
 
@@ -30,9 +30,9 @@ SocketIOClient* socket = [[SocketIOClient alloc] initWithSocketURL:@"localhost:8
 [socket on: @"connect" callback: ^(NSArray* data, void (^ack)(NSArray*)) {
     NSLog(@"connected");
     [socket emitObjc:@"echo" withItems:@[@"echo test"]];
-    [[socket emitWithAckObjc:@"ackack" withItems:@[@"test"]] onAck:0 withCallback:^(NSArray* data) {
-        NSLog(@"Got data");
-    }];
+    [socket emitWithAckObjc:@"connect" withItems:@[@1]](10, ^(NSArray* data) {
+        NSLog(@"Got ack");
+    });
 }];
 
 ```

SwiftIO/SocketAckHandler.swift SwiftIO/SocketAckMap.swiftSwiftIO/SocketAckHandler.swift renamed to SwiftIO/SocketAckMap.swift

@@ -1,9 +1,9 @@
 //
-//  SocketAckHandler.swift
-//  Socket.IO-Swift
+//  SocketAckMap.swift
+//  SocketIO-Swift
+//
+//  Created by Erik Little on 4/3/15.
 //
-//  Created by Erik Little on 2/14/15.
-
 //  Permission is hereby granted, free of charge, to any person obtaining a copy
 //  of this software and associated documentation files (the "Software"), to deal
 //  in the Software without restriction, including without limitation the rights
@@ -25,45 +25,35 @@
 import Foundation
 
 public typealias AckCallback = (NSArray?) -> Void
+public typealias OnAckCallback = (timeout:UInt64, callback:AckCallback) -> Void
 
-@objc public class SocketAckHandler {
-    let ackNum:Int!
-    let event:String!
-    var acked = false
-    var callback:AckCallback?
-    weak var socket:SocketIOClient?
-
+struct SocketAckMap {
+    private var acks = [Int: AckCallback]()
+    private var waiting = [Int: Bool]()
 
-    init(event:String, ackNum:Int = 0, socket:SocketIOClient) {
-        self.ackNum = ackNum
-        self.event = event
-        self.socket = socket
+    mutating func addAck(ack:Int, callback:AckCallback) {
+        waiting[ack] = true
+        acks[ack] = callback
     }
 
-    public func onAck(timeout:UInt64, withCallback callback:AckCallback) {
-        self.callback = callback
-        
+    mutating func executeAck(ack:Int, items:[AnyObject]?) {
+        let callback = acks[ack]
+        waiting[ack] = false
 
-        if timeout != 0 {
-            let time = dispatch_time(DISPATCH_TIME_NOW, Int64(timeout * NSEC_PER_SEC))
-            dispatch_after(time, dispatch_get_main_queue()) {[weak self] in
-                if self == nil {
-                    return
-                }
-                
-                if !self!.acked {
-                    self?.executeAck(["No ACK"])
-                    self?.socket?.removeAck(self!)
-                }
-            }
+        dispatch_async(dispatch_get_main_queue()) {
+            callback?(items)
+            return
         }
+        
+        acks.removeValueForKey(ack)
     }
 
-    func executeAck(data:NSArray?) {
-        dispatch_async(dispatch_get_main_queue()) {[weak self, cb = self.callback] in
-            self?.acked = true
-            cb?(data)
-            return
+    mutating func timeoutAck(ack:Int) {
+        if waiting[ack] != nil && waiting[ack]! {
+            acks[ack]?(["NO ACK"])
         }
+        
+        acks.removeValueForKey(ack)
+        waiting.removeValueForKey(ack)
     }
-}
+}

SwiftIO/SocketEngine.swift

@@ -164,18 +164,21 @@ public class SocketEngine: NSObject, WebSocketDelegate {
         self.ws = WebSocket(url: NSURL(string: self.urlWebSocket! + "&sid=\(self.sid)")!)
         self.ws?.queue = self.handleQueue
         self.ws?.delegate = self
-                
+        
         if connect {
             self.ws?.connect()
         }
     }
 
     private func doFastUpgrade() {
-        self.sendWebSocketMessage("", withType: PacketType.UPGRADE)
-        self._websocket = true
-        self._polling = false
-        self.fastUpgrade = false
-        self.flushProbeWait()
+        dispatch_async(self.emitQueue) {[weak self] in
+            self?.sendWebSocketMessage("", withType: PacketType.UPGRADE)
+            self?._websocket = true
+            self?._polling = false
+            self?.fastUpgrade = false
+            self?.probing = false
+            self?.flushProbeWait()
+        }
     }
 
     private func doPoll() {
@@ -229,7 +232,7 @@ public class SocketEngine: NSObject, WebSocketDelegate {
     }
 
     private func flushProbeWait() {
-        // println("flushing probe wait")
+        // NSLog("flushing probe wait")
         dispatch_async(self.emitQueue) {[weak self] in
             if self == nil {
                 return
@@ -240,6 +243,7 @@ public class SocketEngine: NSObject, WebSocketDelegate {
             }
 
             self?.probeWait.removeAll(keepCapacity: false)
+            // NSLog("waiting for post after flush probe: \(self!.postWait.count)")
         }
     }
 
@@ -269,11 +273,14 @@ public class SocketEngine: NSObject, WebSocketDelegate {
         let postData = postStr.dataUsingEncoding(NSUTF8StringEncoding,
             allowLossyConversion: false)!
 
-        // NSLog("posting: \(postStr)")
         req.HTTPBody = postData
         req.setValue(String(postData.length), forHTTPHeaderField: "Content-Length")
 
         self.waitingForPost = true
+        
+        // NSLog("posting: \(postStr)")
+        // NSLog("Posting with WS status of: \(self.websocket)")
+        
         self.session.dataTaskWithRequest(req) {[weak self] data, res, err in
             if self == nil {
                 return
@@ -284,9 +291,13 @@ public class SocketEngine: NSObject, WebSocketDelegate {
 
             self?.waitingForPost = false
             dispatch_async(self!.emitQueue) {
-                self?.flushWaitingForPost()
-                self?.doPoll()
-                return
+                if self!.fastUpgrade {
+                    self?.doFastUpgrade()
+                    return
+                } else {
+                    self?.flushWaitingForPost()
+                    self?.doPoll()
+                }
             }}.resume()
     }
 
@@ -517,6 +528,8 @@ public class SocketEngine: NSObject, WebSocketDelegate {
         self.write("", withType: PacketType.PING, withData: nil)
     }
 
+    /// Send polling message.
+    /// Only call on emitQueue
     private func sendPollMessage(var msg:String, withType type:PacketType,
         datas:ContiguousArray<NSData>? = nil) {
             // println("Sending poll: \(msg) as type: \(type.rawValue)")
@@ -538,6 +551,8 @@ public class SocketEngine: NSObject, WebSocketDelegate {
             }
     }
 
+    /// Send message on WebSockets
+    /// Only call on emitQueue
     private func sendWebSocketMessage(str:String, withType type:PacketType,
         datas:ContiguousArray<NSData>? = nil) {
             // println("Sending ws: \(str) as type: \(type.rawValue)")
@@ -569,9 +584,9 @@ public class SocketEngine: NSObject, WebSocketDelegate {
 
     private func upgradeTransport() {
         if self.websocketConnected {
+            // NSLog("Doing fast upgrade")
             // Do a fast upgrade
             self.fastUpgrade = true
-            self.probing = false
             self.sendPollMessage("", withType: PacketType.NOOP)
         }
     }
@@ -591,7 +606,7 @@ public class SocketEngine: NSObject, WebSocketDelegate {
             }
         }
     }
-
+    
     // Delagate methods
 
     public func websocketDidConnect(socket:WebSocket) {