Merge pull request #1 from socketio/master

sync

README.md

@@ -6,6 +6,7 @@ Socket.IO-client for iOS/OS X.
 ##Example
 ```swift
 import SocketIO
+
 let socket = SocketIOClient(socketURL: URL(string: "http://localhost:8080")!, config: [.log(true), .forcePolling(true)])
 
 socket.on("connect") {data, ack in
@@ -14,7 +15,7 @@ socket.on("connect") {data, ack in
 
 socket.on("currentAmount") {data, ack in
     if let cur = data[0] as? Double {
-        socket.emitWithAck("canUpdate", cur)(0) {data in
+        socket.emitWithAck("canUpdate", cur).timingOut(after: 0) {data in
             socket.emit("update", ["amount": cur + 2.50])
         }
 
@@ -38,9 +39,9 @@ SocketIOClient* socket = [[SocketIOClient alloc] initWithSocketURL:url config:@{
 [socket on:@"currentAmount" callback:^(NSArray* data, SocketAckEmitter* ack) {
     double cur = [[data objectAtIndex:0] floatValue];
 
-    [socket emitWithAck:@"canUpdate" with:@[@(cur)]](0, ^(NSArray* data) {
+    [[socket emitWithAck:@"canUpdate" with:@[@(cur)] timingOutAfter:0 callback:^(NSArray* data) {
         [socket emit:@"update" withItems:@[@{@"amount": @(cur + 2.50)}]];
-    });
+    }];
 
     [ack with:@[@"Got your currentAmount, ", @"dude"]];
 }];
@@ -94,7 +95,7 @@ Carthage
 -----------------
 Add this line to your `Cartfile`:
 ```
-github "socketio/socket.io-client-swift" ~> 8.0.2 # Or latest version
+github "socketio/socket.io-client-swift" ~> 8.1.0 # Or latest version
 ```
 
 Run `carthage update --platform ios,macosx`.
@@ -107,7 +108,7 @@ Create `Podfile` and add `pod 'Socket.IO-Client-Swift'`:
 use_frameworks!
 
 target 'YourApp' do
-    pod 'Socket.IO-Client-Swift', '~> 8.0.2' # Or latest version
+    pod 'Socket.IO-Client-Swift', '~> 8.1.0' # Or latest version
 end
 ```
 
@@ -136,7 +137,7 @@ CocoaSeeds
 Add this line to your `Seedfile`:
 
 ```
-github "socketio/socket.io-client-swift", "v8.0.2", :files => "Source/*.swift" # Or latest version
+github "socketio/socket.io-client-swift", "v8.1.0", :files => "Source/*.swift" # Or latest version
 ```
 
 Run `seed install`.
@@ -182,8 +183,8 @@ Methods
 3. `onAny(callback:((event: String, items: AnyObject?)) -> Void)` - Adds a handler for all events. It will be called on any received event.
 4. `emit(_ event: String, _ items: AnyObject...)` - Sends a message. Can send multiple items.
 5. `emit(_ event: String, withItems items: [AnyObject])` - `emit` for Objective-C
-6. `emitWithAck(_ event: String, _ items: AnyObject...) -> (timeoutAfter: UInt64, callback: (NSArray?) -> Void) -> Void` - Sends a message that requests an acknowledgement from the server. Returns a function which you can use to add a handler. See example. Note: The message is not sent until you call the returned function.
+6. `emitWithAck(_ event: String, _ items: AnyObject...) -> OnAckCallback` - Sends a message that requests an acknowledgement from the server. Returns an object which you can use to add a handler. See example. Note: The message is not sent until you call timingOut(after:) on the returned object.
-7. `emitWithAck(_ event: String, withItems items: [AnyObject]) -> (UInt64, (NSArray?) -> Void) -> Void` - `emitWithAck` for Objective-C. Note: The message is not sent until you call the returned function.
+7. `emitWithAck(_ event: String, withItems items: [AnyObject]) -> OnAckCallback` - `emitWithAck` for Objective-C. Note: The message is not sent until you call timingOutAfter on the returned object.
 8. `connect()` - Establishes a connection to the server. A "connect" event is fired upon successful connection.
 9. `connect(timeoutAfter timeoutAfter: Int, withTimeoutHandler handler: (() -> Void)?)` - Connect to the server. If it isn't connected after timeoutAfter seconds, the handler is called.
 10. `disconnect()` - Closes the socket. Reopening a disconnected socket is not fully tested.
@@ -205,5 +206,7 @@ Client Events
 ##Detailed Example
 A more detailed example can be found [here](https://github.com/nuclearace/socket.io-client-swift-example)
 
+An example using the Swift Package Manager can be found [here](https://github.com/nuclearace/socket.io-client-swift-spm-example)
+
 ##License
 MIT

Socket.IO-Client-Swift.podspec

@@ -1,7 +1,7 @@
 Pod::Spec.new do |s|
   s.name         = "Socket.IO-Client-Swift"
   s.module_name  = "SocketIO"
-  s.version      = "8.0.2"
+  s.version      = "8.1.0"
   s.summary      = "Socket.IO-client for iOS and OS X"
   s.description  = <<-DESC
                    Socket.IO-client for iOS and OS X.
@@ -14,7 +14,7 @@ Pod::Spec.new do |s|
   s.ios.deployment_target = '8.0'
   s.osx.deployment_target = '10.10'
   s.tvos.deployment_target = '9.0'
-  s.source       = { :git => "https://github.com/socketio/socket.io-client-swift.git", :tag => 'v8.0.2' }
+  s.source       = { :git => "https://github.com/socketio/socket.io-client-swift.git", :tag => 'v8.1.0' }
   s.source_files  = "Source/**/*.swift"
   s.requires_arc = true
   # s.dependency 'Starscream', '~> 0.9' # currently this repo includes Starscream swift files

SocketIO-MacTests/SocketObjectiveCTest.m

@@ -21,7 +21,7 @@
 - (void)setUp {
     [super setUp];
     NSURL* url = [[NSURL alloc] initWithString:@"http://localhost"];
-    self.socket = [[SocketIOClient alloc] initWithSocketURL:url config:nil];
+    self.socket = [[SocketIOClient alloc] initWithSocketURL:url config:@{@"log": @YES, @"forcePolling": @YES}];
 }
 
 - (void)testOnSyntax {
@@ -35,9 +35,9 @@
 }
 
 - (void)testEmitWithAckSyntax {
-    [self.socket emitWithAck:@"testAckEmit" with:@[@YES]](0, ^(NSArray* data) {
+    [[self.socket emitWithAck:@"testAckEmit" with:@[@YES]] timingOutAfter:0 callback:^(NSArray* data) {
         
-    });
+    }];
 }
 
 - (void)testOffSyntax {

SocketIO-MacTests/SocketSideEffectTest.swift

@@ -25,20 +25,20 @@ class SocketSideEffectTest: XCTestCase {
     }
     
     func testFirstAck() {
-        socket.emitWithAck("test")(0) {data in}
+        socket.emitWithAck("test").timingOut(after: 0) {data in}
         XCTAssertEqual(socket.currentAck, 0)
     }
     
     func testSecondAck() {
-        socket.emitWithAck("test")(0) {data in}
+        socket.emitWithAck("test").timingOut(after: 0) {data in}
-        socket.emitWithAck("test")(0) {data in}
+        socket.emitWithAck("test").timingOut(after: 0) {data in}
         
         XCTAssertEqual(socket.currentAck, 1)
     }
     
     func testHandleAck() {
         let expect = expectation(description: "handled ack")
-        socket.emitWithAck("test")(0) {data in
+        socket.emitWithAck("test").timingOut(after: 0) {data in
             XCTAssertEqual(data[0] as? String, "hello world")
             expect.fulfill()
         }
@@ -49,7 +49,7 @@ class SocketSideEffectTest: XCTestCase {
     
     func testHandleAck2() {
         let expect = expectation(description: "handled ack2")
-        socket.emitWithAck("test")(0) {data in
+        socket.emitWithAck("test").timingOut(after: 0) {data in
             XCTAssertTrue(data.count == 2, "Wrong number of ack items")
             expect.fulfill()
         }

Source/SSLSecurity.swift

@@ -4,7 +4,7 @@
 //  Starscream
 //
 //  Created by Dalton Cherry on 5/16/15.
-//  Copyright (c) 2014-2015 Dalton Cherry.
+//  Copyright (c) 2014-2016 Dalton Cherry.
 //
 //  Licensed under the Apache License, Version 2.0 (the "License");
 //  you may not use this file except in compliance with the License.
@@ -19,11 +19,14 @@
 //  limitations under the License.
 //
 //////////////////////////////////////////////////////////////////////////////////////////////////
-
 import Foundation
 import Security
 
-public class SSLCert : NSObject {
+public protocol SSLTrustValidator {
+    func isValid(_ trust: SecTrust, domain: String?) -> Bool
+}
+
+open class SSLCert {
     var certData: Data?
     var key: SecKey?
     
@@ -50,7 +53,7 @@ public class SSLCert : NSObject {
     }
 }
 
-public class SSLSecurity : NSObject {
+open class SSLSecurity : SSLTrustValidator {
     public var validatedDN = true //should the domain name be validated?
     
     var isReady = false //is the key processing done?
@@ -82,7 +85,7 @@ public class SSLSecurity : NSObject {
     /**
      Designated init
      
-     - parameter keys: is the certificates or public keys to use
+     - parameter certs: is the certificates or public keys to use
      - parameter usePublicKeys: is to specific if the publicKeys or certificates should be used for SSL pinning validation
      
      - returns: a representation security object to be used with
@@ -90,8 +93,6 @@ public class SSLSecurity : NSObject {
     public init(certs: [SSLCert], usePublicKeys: Bool) {
         self.usePublicKeys = usePublicKeys
         
-        super.init()
-        
         if self.usePublicKeys {
             DispatchQueue.global(qos: .default).async {
                 let pubKeys = certs.reduce([SecKey]()) { (pubKeys: [SecKey], cert: SSLCert) -> [SecKey] in

Source/SocketAckEmitter.swift

@@ -45,3 +45,38 @@ public final class SocketAckEmitter : NSObject {
         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._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: time) {
+            socket.ackHandlers.timeoutAck(self.ackNumber, onQueue: socket.handleQueue)
+        }
+    }
+    
+}

Source/SocketIOClient.swift

@@ -46,21 +46,22 @@ public final class SocketIOClient : NSObject, SocketEngineClient, SocketParsable
     public var reconnects = true
     public var reconnectWait = 10
 
-    private let ackQueue = DispatchQueue(label: "com.socketio.ackQueue", attributes: [])
-    private let emitQueue = DispatchQueue(label: "com.socketio.emitQueue", attributes: [])
     private let logType = "SocketIOClient"
-    private let parseQueue = DispatchQueue(label: "com.socketio.parseQueue", attributes: [])
+    private let parseQueue = DispatchQueue(label: "com.socketio.parseQueue")
 
     private var anyHandler: ((SocketAnyEvent) -> Void)?
     private var currentReconnectAttempt = 0
     private var handlers = [SocketEventHandler]()
-    private var ackHandlers = SocketAckManager()
     private var reconnecting = false
 
     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? {
@@ -148,7 +149,7 @@ public final class SocketIOClient : NSObject, SocketEngineClient, SocketParsable
         
         guard timeoutAfter != 0 else { return }
 
-        let time = DispatchTime.now() + Double(Int64(timeoutAfter) * Int64(NSEC_PER_SEC)) / Double(NSEC_PER_SEC)
+        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 }
@@ -163,24 +164,7 @@ public final class SocketIOClient : NSObject, SocketEngineClient, SocketParsable
     private func createOnAck(_ items: [Any]) -> OnAckCallback {
         currentAck += 1
         
-        return {[weak self, ack = currentAck] timeout, callback in
+        return OnAckCallback(ackNumber: currentAck, items: items, socket: self)
-            guard let this = self else { return }
-
-            this.ackQueue.sync() {
-                this.ackHandlers.addAck(ack, callback: callback)
-            }
-            
-            
-            this._emit(items, ack: ack)
-            
-            if timeout != 0 {
-                let time = DispatchTime.now() + Double(Int64(timeout * NSEC_PER_SEC)) / Double(NSEC_PER_SEC)
-                
-                this.handleQueue.asyncAfter(deadline: time) {
-                    this.ackHandlers.timeoutAck(ack, onQueue: this.handleQueue)
-                }
-            }
-        }
     }
 
     func didConnect() {
@@ -238,7 +222,7 @@ public final class SocketIOClient : NSObject, SocketEngineClient, SocketParsable
         return createOnAck([event] + items)
     }
 
-    private func _emit(_ data: [Any], ack: Int? = nil) {
+    func _emit(_ data: [Any], ack: Int? = nil) {
         emitQueue.async {
             guard self.status == .connected else {
                 self.handleEvent("error", data: ["Tried emitting when not connected"], isInternalMessage: true)
@@ -269,7 +253,9 @@ public final class SocketIOClient : NSObject, SocketEngineClient, SocketParsable
     }
 
     public func engineDidClose(reason: String) {
-        waitingPackets.removeAll()
+        parseQueue.sync {
+            self.waitingPackets.removeAll()
+        }
         
         if status != .disconnected {
             status = .notConnected

Source/SocketPacket.swift

@@ -87,25 +87,18 @@ struct SocketPacket {
     }
     
     private func completeMessage(_ message: String) -> String {
-        let restOfMessage: String
-        
         if data.count == 0 {
             return message + "[]"
         }
         
-        do {
+        guard let jsonSend = try? data.toJSON(), let jsonString = String(data: jsonSend, encoding: .utf8) else {
-            let jsonSend = try data.toJSON()
-            guard let jsonString = String(data: jsonSend, encoding: .utf8) else { return message + "[]" }
-            
-            restOfMessage = jsonString
-        } catch {
             DefaultSocketLogger.Logger.error("Error creating JSON object in SocketPacket.completeMessage",
                                              type: SocketPacket.logType)
             
-            restOfMessage = "[]"
+            return message + "[]"
         }
         
-        return message + restOfMessage
+        return message + jsonString
     }
     
     private func createPacketString() -> String {
@@ -133,11 +126,11 @@ struct SocketPacket {
     // binary data
     private func _fillInPlaceholders(_ object: Any) -> Any {
         switch object {
-        case let dict as [String: Any]:
+        case let dict as JSON:
             if dict["_placeholder"] as? Bool ?? false {
                 return binary[dict["num"] as! Int]
             } else {
-                return dict.reduce([String: Any](), {cur, keyValue in
+                return dict.reduce(JSON(), {cur, keyValue in
                     var cur = cur
                     
                     cur[keyValue.0] = _fillInPlaceholders(keyValue.1)
@@ -181,7 +174,7 @@ extension SocketPacket {
 private extension SocketPacket {
     // Recursive function that looks for NSData in collections
     static func shred(_ data: Any, binary: inout [Data]) -> Any {
-        let placeholder = ["_placeholder": true, "num": binary.count] as [String : Any]
+        let placeholder = ["_placeholder": true, "num": binary.count] as JSON
         
         switch data {
         case let bin as Data:
@@ -190,8 +183,8 @@ private extension SocketPacket {
             return placeholder
         case let arr as [Any]:
             return arr.map({shred($0, binary: &binary)})
-        case let dict as [String: Any]:
+        case let dict as JSON:
-            return dict.reduce([String: Any](), {cur, keyValue in
+            return dict.reduce(JSON(), {cur, keyValue in
                 var mutCur = cur
                 
                 mutCur[keyValue.0] = shred(keyValue.1, binary: &binary)

Source/SocketTypes.swift

@@ -41,8 +41,8 @@ extension String : SocketData {}
 
 public typealias AckCallback = ([Any]) -> Void
 public typealias NormalCallback = ([Any], SocketAckEmitter) -> Void
-public typealias OnAckCallback = (_ timeoutAfter: UInt64, _ callback: @escaping AckCallback) -> Void
 
+typealias JSON = [String: Any]
 typealias Probe = (msg: String, type: SocketEnginePacketType, data: [Data])
 typealias ProbeWaitQueue = [Probe]
 

Source/WebSocket.swift

@@ -3,7 +3,7 @@
 //  Websocket.swift
 //
 //  Created by Dalton Cherry on 7/16/14.
-//  Copyright (c) 2014-2015 Dalton Cherry.
+//  Copyright (c) 2014-2016 Dalton Cherry.
 //
 //  Licensed under the Apache License, Version 2.0 (the "License");
 //  you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@
 //  limitations under the License.
 //
 //////////////////////////////////////////////////////////////////////////////////////////////////
-
 import Foundation
 import CoreFoundation
 import Security
@@ -38,7 +37,7 @@ public protocol WebSocketPongDelegate: class {
     func websocketDidReceivePong(socket: WebSocket, data: Data?)
 }
 
-public class WebSocket : NSObject, StreamDelegate {
+open class WebSocket : NSObject, StreamDelegate {
     
     enum OpCode : UInt8 {
         case continueFrame = 0x0
@@ -77,7 +76,6 @@ public class WebSocket : NSObject, StreamDelegate {
     var optionalProtocols: [String]?
     
     // MARK: - Constants
-    
     let headerWSUpgradeName     = "Upgrade"
     let headerWSUpgradeValue    = "websocket"
     let headerWSHostName        = "Host"
@@ -108,7 +106,6 @@ public class WebSocket : NSObject, StreamDelegate {
     }
     
     // MARK: - Delegates
-    
     /// Responds to callback about new messages coming in over the WebSocket
     /// and also connection/disconnect messages.
     public weak var delegate: WebSocketDelegate?
@@ -118,7 +115,6 @@ public class WebSocket : NSObject, StreamDelegate {
     
     
     // MARK: - Block based API.
-    
     public var onConnect: ((Void) -> Void)?
     public var onDisconnect: ((NSError?) -> Void)?
     public var onText: ((String) -> Void)?
@@ -128,7 +124,7 @@ public class WebSocket : NSObject, StreamDelegate {
     public var headers = [String: String]()
     public var voipEnabled = false
     public var disableSSLCertValidation = false
-    public var security: SSLSecurity?
+    public var security: SSLTrustValidator?
     public var enabledSSLCipherSuites: [SSLCipherSuite]?
     public var origin: String?
     public var timeout = 5
@@ -139,7 +135,6 @@ public class WebSocket : NSObject, StreamDelegate {
     public var currentURL: URL { return url }
     
     // MARK: - Private
-    
     private var url: URL
     private var inputStream: InputStream?
     private var outputStream: OutputStream?
@@ -198,7 +193,8 @@ public class WebSocket : NSObject, StreamDelegate {
     public func disconnect(forceTimeout: TimeInterval? = nil, closeCode: UInt16 = CloseCode.normal.rawValue) {
         switch forceTimeout {
         case .some(let seconds) where seconds > 0:
-            callbackQueue.asyncAfter(deadline: DispatchTime.now() + Double(Int64(seconds * Double(NSEC_PER_SEC))) / Double(NSEC_PER_SEC)) { [weak self] in
+            let milliseconds = Int(seconds * 1_000)
+            callbackQueue.asyncAfter(deadline: .now() + .milliseconds(milliseconds)) { [weak self] in
                 self?.disconnectStream(nil)
             }
             fallthrough
@@ -213,7 +209,7 @@ public class WebSocket : NSObject, StreamDelegate {
     /**
      Write a string to the websocket. This sends it as a text frame.
      If you supply a non-nil completion block, I will perform it when the write completes.
-     - parameter str:        The string to write.
+     - parameter string:        The string to write.
      - parameter completion: The (optional) completion handler.
      */
     public func write(string: String, completion: (() -> ())? = nil) {
@@ -305,7 +301,6 @@ public class WebSocket : NSObject, StreamDelegate {
     private func initStreamsWithData(_ data: Data, _ port: Int) {
         //higher level API we will cut over to at some point
         //NSStream.getStreamsToHostWithName(url.host, port: url.port.integerValue, inputStream: &inputStream, outputStream: &outputStream)
-        
         var readStream: Unmanaged<CFReadStream>?
         var writeStream: Unmanaged<CFWriteStream>?
         let h = url.host! as NSString
@@ -318,13 +313,6 @@ public class WebSocket : NSObject, StreamDelegate {
         if supportedSSLSchemes.contains(url.scheme!) {
             inStream.setProperty(StreamSocketSecurityLevel.negotiatedSSL as AnyObject, forKey: Stream.PropertyKey.socketSecurityLevelKey)
             outStream.setProperty(StreamSocketSecurityLevel.negotiatedSSL as AnyObject, forKey: Stream.PropertyKey.socketSecurityLevelKey)
-        } else {
-            certValidated = true //not a https session, so no need to check SSL pinning
-        }
-        if voipEnabled {
-            inStream.setProperty(StreamNetworkServiceTypeValue.voIP as AnyObject, forKey: Stream.PropertyKey.networkServiceType)
-            outStream.setProperty(StreamNetworkServiceTypeValue.voIP as AnyObject, forKey: Stream.PropertyKey.networkServiceType)
-        }
             if disableSSLCertValidation {
                 let settings: [NSObject: NSObject] = [kCFStreamSSLValidatesCertificateChain: NSNumber(value: false), kCFStreamSSLPeerName: kCFNull]
                 inStream.setProperty(settings, forKey: kCFStreamPropertySSLSettings as Stream.PropertyKey)
@@ -347,6 +335,13 @@ public class WebSocket : NSObject, StreamDelegate {
                     }
                 }
             }
+        } else {
+            certValidated = true //not a https session, so no need to check SSL pinning
+        }
+        if voipEnabled {
+            inStream.setProperty(StreamNetworkServiceTypeValue.voIP as AnyObject, forKey: Stream.PropertyKey.networkServiceType)
+            outStream.setProperty(StreamNetworkServiceTypeValue.voIP as AnyObject, forKey: Stream.PropertyKey.networkServiceType)
+        }
         
         CFReadStreamSetDispatchQueue(inStream, WebSocket.sharedWorkQueue)
         CFWriteStreamSetDispatchQueue(outStream, WebSocket.sharedWorkQueue)
@@ -358,7 +353,7 @@ public class WebSocket : NSObject, StreamDelegate {
         self.mutex.unlock()
         
         let bytes = UnsafeRawPointer((data as NSData).bytes).assumingMemoryBound(to: UInt8.self)
-        var out = timeout * 1000000 // wait 5 seconds before giving up
+        var out = timeout * 1_000_000 // wait 5 seconds before giving up
         writeQueue.addOperation { [weak self] in
             while !outStream.hasSpaceAvailable {
                 usleep(100) // wait until the socket is ready
@@ -380,9 +375,9 @@ public class WebSocket : NSObject, StreamDelegate {
      */
     public func stream(_ aStream: Stream, handle eventCode: Stream.Event) {
         if let sec = security, !certValidated && [.hasBytesAvailable, .hasSpaceAvailable].contains(eventCode) {
-            let trust = aStream.property(forKey: kCFStreamPropertySSLPeerTrust as Stream.PropertyKey) as AnyObject
+            let trust = aStream.property(forKey: kCFStreamPropertySSLPeerTrust as Stream.PropertyKey) as! SecTrust
             let domain = aStream.property(forKey: kCFStreamSSLPeerName as Stream.PropertyKey) as? String
-            if sec.isValid(trust as! SecTrust, domain: domain) {
+            if sec.isValid(trust, domain: domain) {
                 certValidated = true
             } else {
                 let error = errorWithDetail("Invalid SSL certificate", code: 1)
@@ -439,7 +434,6 @@ public class WebSocket : NSObject, StreamDelegate {
         let buf = NSMutableData(capacity: BUFFER_MAX)
         let buffer = UnsafeMutableRawPointer(mutating: buf!.bytes).assumingMemoryBound(to: UInt8.self)
         let length = inputStream!.read(buffer, maxLength: BUFFER_MAX)
-        
         guard length > 0 else { return }
         var process = false
         if inputQueue.count == 0 {
@@ -635,34 +629,22 @@ public class WebSocket : NSObject, StreamDelegate {
                 writeError(errCode)
                 return emptyBuffer
             }
+            var closeCode = CloseCode.normal.rawValue
             if receivedOpcode == .connectionClose {
-                var code = CloseCode.normal.rawValue
                 if payloadLen == 1 {
-                    code = CloseCode.protocolError.rawValue
+                    closeCode = CloseCode.protocolError.rawValue
                 } else if payloadLen > 1 {
-                    code = WebSocket.readUint16(baseAddress, offset: offset)
+                    closeCode = WebSocket.readUint16(baseAddress, offset: offset)
-                    if code < 1000 || (code > 1003 && code < 1007) || (code > 1011 && code < 3000) {
+                    if closeCode < 1000 || (closeCode > 1003 && closeCode < 1007) || (closeCode > 1011 && closeCode < 3000) {
-                        code = CloseCode.protocolError.rawValue
+                        closeCode = CloseCode.protocolError.rawValue
-                    }
-                    offset += 2
-                }
-                var closeReason = "connection closed by server"
-                if payloadLen > 2 {
-                    let len = Int(payloadLen - 2)
-                    if len > 0 {
-                        let bytes = baseAddress + offset
-                        if let customCloseReason = String(data: Data(bytes: bytes, count: len), encoding: .utf8) {
-                            closeReason = customCloseReason
-                        } else {
-                            code = CloseCode.protocolError.rawValue
                     }
                 }
-                }
+                if payloadLen < 2 {
-                doDisconnect(errorWithDetail(closeReason, code: code))
+                    doDisconnect(errorWithDetail("connection closed by server", code: closeCode))
-                writeError(code)
+                    writeError(closeCode)
                     return emptyBuffer
                 }
-            if isControlFrame && payloadLen > 125 {
+            } else if isControlFrame && payloadLen > 125 {
                 writeError(CloseCode.protocolError.rawValue)
                 return emptyBuffer
             }
@@ -687,8 +669,24 @@ public class WebSocket : NSObject, StreamDelegate {
                 len = 0
                 data = Data()
             } else {
+                if receivedOpcode == .connectionClose && len > 0 {
+                    let size = MemoryLayout<UInt16>.size
+                    offset += size
+                    len -= UInt64(size)
+                }
                 data = Data(bytes: baseAddress+offset, count: Int(len))
             }
+            if receivedOpcode == .connectionClose {
+                var closeReason = "connection closed by server"
+                if let customCloseReason = String(data: data, encoding: .utf8) {
+                    closeReason = customCloseReason
+                } else {
+                    closeCode = CloseCode.protocolError.rawValue
+                }
+                doDisconnect(errorWithDetail(closeReason, code: closeCode))
+                writeError(closeCode)
+                return emptyBuffer
+            }
             if receivedOpcode == .pong {
                 if canDispatch {
                     callbackQueue.async { [weak self] in
@@ -902,7 +900,6 @@ public class WebSocket : NSObject, StreamDelegate {
     }
     
     // MARK: - Deinit
-    
     deinit {
         mutex.lock()
         readyToWrite = false