Use starscream for websockets. Execute emitAcks on the main queue.
This commit is contained in:
Erik
parent
2e5a85a3f2
commit
8365ded215
@ -7,9 +7,7 @@ For Swift 1.2 use the 1.2 branch.
|
||||
|
||||
Installation
|
||||
============
|
||||
1. Requires linking [SocketRocket](https://github.com/square/SocketRocket) against your xcode project. (Be sure to link the [frameworks](https://github.com/square/SocketRocket#framework-dependencies) required by SocketRocket)
|
||||
2. Create a bridging header for SocketRocket
|
||||
3. Copy the SwiftIO folder into your xcode project
|
||||
1. Copy the SwiftIO folder into your Xcode project!
|
||||
|
||||
API
|
||||
===
|
||||
|
||||
@ -39,4 +39,11 @@ class SocketAckHandler {
|
||||
func onAck(callback:AckCallback) {
|
||||
self.callback = callback
|
||||
}
|
||||
|
||||
func executeAck(data:NSArray?) {
|
||||
dispatch_async(dispatch_get_main_queue()) {[cb = self.callback] in
|
||||
cb?(data)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -46,7 +46,7 @@ private enum PacketType: String {
|
||||
case NOOP = "6"
|
||||
}
|
||||
|
||||
class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
class SocketEngine: NSObject, WebSocketDelegate {
|
||||
unowned let client:SocketIOClient
|
||||
private let workQueue = NSOperationQueue()
|
||||
private let emitQueue = dispatch_queue_create(
|
||||
@ -77,7 +77,7 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
var websocket:Bool {
|
||||
return self._websocket
|
||||
}
|
||||
var ws:SRWebSocket?
|
||||
var ws:WebSocket?
|
||||
|
||||
init(client:SocketIOClient, forcePolling:Bool = false) {
|
||||
self.client = client
|
||||
@ -165,7 +165,11 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
if let str = NSString(data: data, encoding: NSUTF8StringEncoding) as? String {
|
||||
// println(str)
|
||||
|
||||
dispatch_async(self?.parseQueue) {[weak self] in
|
||||
dispatch_async(self!.parseQueue) {[weak self] in
|
||||
if self == nil {
|
||||
return
|
||||
}
|
||||
|
||||
self?.parsePollingMessage(str)
|
||||
return
|
||||
}
|
||||
@ -224,12 +228,10 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
self.waitingForPost = true
|
||||
|
||||
self.session.dataTaskWithRequest(req) {[weak self] data, res, err in
|
||||
if err != nil {
|
||||
if self!.polling {
|
||||
self?.handlePollingFailed(err)
|
||||
}
|
||||
if self == nil {
|
||||
return
|
||||
} else if self == nil {
|
||||
} else if err != nil && self!.polling {
|
||||
self?.handlePollingFailed(err)
|
||||
return
|
||||
}
|
||||
|
||||
@ -245,7 +247,7 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
// Send them raw
|
||||
private func flushWaitingForPostToWebSocket() {
|
||||
for msg in self.postWait {
|
||||
self.ws?.send(msg)
|
||||
self.ws?.writeString(msg)
|
||||
}
|
||||
|
||||
self.postWait.removeAll(keepCapacity: true)
|
||||
@ -256,7 +258,7 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
private func handlePollingFailed(reason:NSError?) {
|
||||
if !self.client.reconnecting {
|
||||
self.connected = false
|
||||
self.ws?.close()
|
||||
self.ws?.disconnect()
|
||||
self.pingTimer?.invalidate()
|
||||
self.waitingForPoll = false
|
||||
self.waitingForPost = false
|
||||
@ -310,10 +312,10 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
self?.sid = sid
|
||||
|
||||
if !self!.forcePolling {
|
||||
self?.ws = SRWebSocket(URL:
|
||||
NSURL(string: urlWebSocket + "&sid=\(self!.sid)")!)
|
||||
self?.ws = WebSocket(url: NSURL(string: urlWebSocket + "&sid=\(self!.sid)")!)
|
||||
self?.ws?.queue = self?.handleQueue
|
||||
self?.ws?.delegate = self
|
||||
self?.ws?.open()
|
||||
self?.ws?.connect()
|
||||
}
|
||||
} else {
|
||||
NSLog("Error handshaking")
|
||||
@ -377,8 +379,9 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
// Be sure to capture the value of the msg
|
||||
dispatch_async(self.handleQueue) {[weak self, msg] in
|
||||
fixSwift = msg
|
||||
self?.parseEngineMessage(fixSwift)
|
||||
return
|
||||
if fixSwift is String {
|
||||
self?.parseEngineMessage(fixSwift as String)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -388,16 +391,14 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
}
|
||||
}
|
||||
|
||||
private func parseEngineMessage(message:AnyObject?) {
|
||||
private func parseEngineData(data:NSData) {
|
||||
self.client.parseBinaryData(data.subdataWithRange(NSMakeRange(1, data.length - 1)))
|
||||
}
|
||||
|
||||
private func parseEngineMessage(var message:String) {
|
||||
// println(message!)
|
||||
if let data = message as? NSData {
|
||||
// Strip off message type
|
||||
self.client.parseSocketMessage(data.subdataWithRange(NSMakeRange(1, data.length - 1)))
|
||||
return
|
||||
}
|
||||
|
||||
var messageString = message as String
|
||||
var strMessage = RegexMutable(messageString)
|
||||
var strMessage = RegexMutable(message)
|
||||
|
||||
// We should upgrade
|
||||
if strMessage == "3probe" {
|
||||
@ -409,16 +410,16 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
|
||||
if type != PacketType.MESSAGE.rawValue {
|
||||
// TODO Handle other packets
|
||||
if messageString.hasPrefix("b4") {
|
||||
if message.hasPrefix("b4") {
|
||||
// binary in base64 string
|
||||
|
||||
messageString.removeRange(Range<String.Index>(start: messageString.startIndex,
|
||||
end: advance(messageString.startIndex, 2)))
|
||||
message.removeRange(Range<String.Index>(start: message.startIndex,
|
||||
end: advance(message.startIndex, 2)))
|
||||
|
||||
if let data = NSData(base64EncodedString: messageString,
|
||||
if let data = NSData(base64EncodedString: message,
|
||||
options: NSDataBase64DecodingOptions.IgnoreUnknownCharacters) {
|
||||
// println("sending \(data)")
|
||||
self.client.parseSocketMessage(data)
|
||||
self.client.parseBinaryData(data)
|
||||
}
|
||||
|
||||
return
|
||||
@ -427,7 +428,7 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
return
|
||||
}
|
||||
|
||||
if messageString == PacketType.CLOSE.rawValue {
|
||||
if message == PacketType.CLOSE.rawValue {
|
||||
// do nothing
|
||||
return
|
||||
}
|
||||
@ -436,10 +437,10 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
}
|
||||
|
||||
// Remove message type
|
||||
messageString.removeAtIndex(messageString.startIndex)
|
||||
message.removeAtIndex(message.startIndex)
|
||||
// println("sending \(messageString)")
|
||||
|
||||
self.client.parseSocketMessage(messageString)
|
||||
self.client.parseSocketMessage(message)
|
||||
}
|
||||
|
||||
private func probeWebSocket() {
|
||||
@ -480,6 +481,7 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
|
||||
func sendPing() {
|
||||
if self.websocket {
|
||||
self.ws?.writePing(NSData())
|
||||
self.sendWebSocketMessage("", withType: PacketType.PING)
|
||||
} else {
|
||||
self.sendPollMessage("", withType: PacketType.PING)
|
||||
@ -510,13 +512,13 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
|
||||
private func sendWebSocketMessage(str:String, withType type:PacketType, datas:[NSData]? = nil) {
|
||||
// println("Sending: ws: \(str) as type: \(type.rawValue)")
|
||||
self.ws?.send("\(type.rawValue)\(str)")
|
||||
self.ws?.writeString("\(type.rawValue)\(str)")
|
||||
|
||||
if datas != nil {
|
||||
for data in datas! {
|
||||
let (data, nilString) = self.createBinaryDataForSend(data)
|
||||
if data != nil {
|
||||
self.ws?.send(data!)
|
||||
self.ws?.writeData(data!)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -546,25 +548,13 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
}
|
||||
}
|
||||
|
||||
// Called when a message is recieved
|
||||
func webSocket(webSocket:SRWebSocket!, didReceiveMessage message:AnyObject?) {
|
||||
// println(message)
|
||||
|
||||
dispatch_async(self.handleQueue) {[weak self] in
|
||||
self?.parseEngineMessage(message)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Called when the socket is opened
|
||||
func webSocketDidOpen(webSocket:SRWebSocket!) {
|
||||
func websocketDidConnect(socket:WebSocket) {
|
||||
self.websocketConnected = true
|
||||
self.probing = true
|
||||
self.probeWebSocket()
|
||||
}
|
||||
|
||||
// Called when the socket is closed
|
||||
func webSocket(webSocket:SRWebSocket!, didCloseWithCode code:Int, reason:String!, wasClean:Bool) {
|
||||
func websocketDidDisconnect(socket:WebSocket, error:NSError?) {
|
||||
self.websocketConnected = false
|
||||
self.probing = false
|
||||
|
||||
@ -573,24 +563,17 @@ class SocketEngine: NSObject, SRWebSocketDelegate {
|
||||
self.connected = false
|
||||
self._websocket = false
|
||||
self._polling = true
|
||||
self.client.webSocketDidCloseWithCode(code, reason: reason, wasClean: wasClean)
|
||||
self.client.webSocketDidCloseWithCode(1, reason: "Socket Disconnect", wasClean: true)
|
||||
} else {
|
||||
self.flushProbeWait()
|
||||
}
|
||||
}
|
||||
|
||||
// Called when an error occurs.
|
||||
func webSocket(webSocket:SRWebSocket!, didFailWithError error:NSError!) {
|
||||
self.websocketConnected = false
|
||||
self._polling = true
|
||||
self.probing = false
|
||||
|
||||
if self.websocket {
|
||||
self.pingTimer?.invalidate()
|
||||
self.connected = false
|
||||
self.client.webSocketDidFailWithError(error)
|
||||
} else {
|
||||
self.flushProbeWait()
|
||||
}
|
||||
func websocketDidReceiveMessage(socket:WebSocket, text:String) {
|
||||
self.parseEngineMessage(text)
|
||||
}
|
||||
|
||||
func websocketDidReceiveData(socket:WebSocket, data:NSData) {
|
||||
self.parseEngineData(data)
|
||||
}
|
||||
}
|
||||
@ -22,6 +22,8 @@
|
||||
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
// THE SOFTWARE.
|
||||
|
||||
import Foundation
|
||||
|
||||
typealias NormalCallback = (NSArray?, AckEmitter?) -> Void
|
||||
typealias AnyHandler = (event:String, items:AnyObject?)
|
||||
typealias AckEmitter = (AnyObject...) -> Void
|
||||
|
||||
@ -25,7 +25,6 @@
|
||||
import Foundation
|
||||
|
||||
class SocketIOClient {
|
||||
let engine:SocketEngine!
|
||||
let socketURL:NSMutableString!
|
||||
let ackQueue = dispatch_queue_create("ackQueue".cStringUsingEncoding(NSUTF8StringEncoding),
|
||||
DISPATCH_QUEUE_SERIAL)
|
||||
@ -48,6 +47,7 @@ class SocketIOClient {
|
||||
var closed = false
|
||||
var connected = false
|
||||
var connecting = false
|
||||
var engine:SocketEngine?
|
||||
var nsp:String?
|
||||
var reconnects = true
|
||||
var reconnecting = false
|
||||
@ -68,7 +68,6 @@ class SocketIOClient {
|
||||
mutURL = mutURL["https://"] ~= ""
|
||||
|
||||
self.socketURL = mutURL
|
||||
self.reconnectAttempts = -1
|
||||
|
||||
// Set options
|
||||
if opts != nil {
|
||||
@ -78,6 +77,8 @@ class SocketIOClient {
|
||||
|
||||
if let reconnectAttempts = opts!["reconnectAttempts"] as? Int {
|
||||
self.reconnectAttempts = reconnectAttempts
|
||||
} else {
|
||||
self.reconnectAttempts = -1
|
||||
}
|
||||
|
||||
if let reconnectWait = opts!["reconnectWait"] as? Int {
|
||||
@ -91,6 +92,8 @@ class SocketIOClient {
|
||||
if let polling = opts!["forcePolling"] as? Bool {
|
||||
self.forcePolling = polling
|
||||
}
|
||||
} else {
|
||||
self.reconnectAttempts = -1
|
||||
}
|
||||
|
||||
self.engine = SocketEngine(client: self, forcePolling: self.forcePolling)
|
||||
@ -112,7 +115,7 @@ class SocketIOClient {
|
||||
self.closed = false
|
||||
}
|
||||
|
||||
self.engine.open()
|
||||
self.engine?.open()
|
||||
}
|
||||
|
||||
// Connect to the server using params
|
||||
@ -125,7 +128,7 @@ class SocketIOClient {
|
||||
self.params = params
|
||||
self.paramConnect = true
|
||||
|
||||
self.engine.open(opts: params)
|
||||
self.engine?.open(opts: params)
|
||||
}
|
||||
|
||||
func didConnect() {
|
||||
@ -254,11 +257,11 @@ class SocketIOClient {
|
||||
return true
|
||||
} else {
|
||||
if data is NSArray {
|
||||
handler.callback?(data as? NSArray)
|
||||
handler.executeAck(data as? NSArray)
|
||||
} else if data != nil {
|
||||
handler.callback?([data!])
|
||||
handler.executeAck([data!])
|
||||
} else {
|
||||
handler.callback?(nil)
|
||||
handler.executeAck(nil)
|
||||
}
|
||||
|
||||
return false
|
||||
@ -494,77 +497,86 @@ class SocketIOClient {
|
||||
}
|
||||
|
||||
// Parses messages recieved
|
||||
func parseSocketMessage(message:AnyObject?) {
|
||||
if message == nil {
|
||||
return
|
||||
}
|
||||
|
||||
func parseSocketMessage(stringMessage:String) {
|
||||
// println(message!)
|
||||
|
||||
if let stringMessage = message as? String {
|
||||
// Check for successful namepsace connect
|
||||
// Check for successful namepsace connect
|
||||
if self.nsp != nil {
|
||||
if stringMessage == "0/\(self.nsp!)" {
|
||||
self.didConnect()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if stringMessage == "0" {
|
||||
if self.nsp != nil {
|
||||
if stringMessage == "0/\(self.nsp!)" {
|
||||
self.didConnect()
|
||||
return
|
||||
}
|
||||
// Join namespace
|
||||
self.joinNamespace()
|
||||
return
|
||||
} else {
|
||||
// Don't handle as internal because something crazy could happen where
|
||||
// we disconnect before it's handled
|
||||
self.didConnect()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var mutMessage = RegexMutable(stringMessage)
|
||||
|
||||
/**
|
||||
Begin check for message
|
||||
**/
|
||||
let messageGroups = mutMessage["(\\d*)\\/?(\\w*)?,?(\\d*)?(\\[.*\\])?"].groups()
|
||||
|
||||
if messageGroups[1].hasPrefix("2") {
|
||||
var mesNum = messageGroups[1]
|
||||
var ackNum:String
|
||||
var namespace:String?
|
||||
var messagePart:String!
|
||||
|
||||
if messageGroups[3] != "" {
|
||||
ackNum = messageGroups[3]
|
||||
} else {
|
||||
let range = Range<String.Index>(start: mesNum.startIndex, end: advance(mesNum.startIndex, 1))
|
||||
mesNum.replaceRange(range, with: "")
|
||||
ackNum = mesNum
|
||||
}
|
||||
|
||||
if stringMessage == "0" {
|
||||
if self.nsp != nil {
|
||||
// Join namespace
|
||||
self.joinNamespace()
|
||||
return
|
||||
} else {
|
||||
// Don't handle as internal because something crazy could happen where
|
||||
// we disconnect before it's handled
|
||||
self.didConnect()
|
||||
return
|
||||
}
|
||||
namespace = messageGroups[2]
|
||||
messagePart = messageGroups[4]
|
||||
|
||||
if namespace == "" && self.nsp != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var mutMessage = RegexMutable(stringMessage)
|
||||
|
||||
/**
|
||||
Begin check for message
|
||||
**/
|
||||
let messageGroups = mutMessage["(\\d*)\\/?(\\w*)?,?(\\d*)?(\\[.*\\])?"].groups()
|
||||
|
||||
if messageGroups[1].hasPrefix("2") {
|
||||
var mesNum = messageGroups[1]
|
||||
var ackNum:String
|
||||
var namespace:String?
|
||||
var messagePart:String!
|
||||
let messageInternals = RegexMutable(messagePart)["\\[\"(.*?)\",(.*?)?\\]$"].groups()
|
||||
if messageInternals != nil && messageInternals.count > 2 {
|
||||
let event = messageInternals[1]
|
||||
var data:String?
|
||||
|
||||
if messageGroups[3] != "" {
|
||||
ackNum = messageGroups[3]
|
||||
if messageInternals[2] == "" {
|
||||
data = nil
|
||||
} else {
|
||||
let range = Range<String.Index>(start: mesNum.startIndex, end: advance(mesNum.startIndex, 1))
|
||||
mesNum.replaceRange(range, with: "")
|
||||
ackNum = mesNum
|
||||
data = messageInternals[2]
|
||||
}
|
||||
|
||||
namespace = messageGroups[2]
|
||||
messagePart = messageGroups[4]
|
||||
|
||||
if namespace == "" && self.nsp != nil {
|
||||
return
|
||||
}
|
||||
|
||||
let messageInternals = RegexMutable(messagePart)["\\[\"(.*?)\",(.*?)?\\]$"].groups()
|
||||
if messageInternals != nil && messageInternals.count > 2 {
|
||||
let event = messageInternals[1]
|
||||
var data:String?
|
||||
|
||||
if messageInternals[2] == "" {
|
||||
data = nil
|
||||
// It would be nice if socket.io only allowed one thing
|
||||
// per message, but alas, it doesn't.
|
||||
if let parsed:AnyObject = SocketIOClient.parseData(data) {
|
||||
if ackNum == "" {
|
||||
self.handleEvent(event, data: parsed)
|
||||
} else {
|
||||
data = messageInternals[2]
|
||||
self.currentAck = ackNum.toInt()!
|
||||
self.handleEvent(event, data: parsed, isInternalMessage: false,
|
||||
wantsAck: ackNum.toInt(), withAckType: 3)
|
||||
}
|
||||
|
||||
// It would be nice if socket.io only allowed one thing
|
||||
// per message, but alas, it doesn't.
|
||||
if let parsed:AnyObject = SocketIOClient.parseData(data) {
|
||||
return
|
||||
} else if let strData = data {
|
||||
// There are multiple items in the message
|
||||
// Turn it into a String and run it through
|
||||
// parseData to try and get an array.
|
||||
let asArray = "[\(strData)]"
|
||||
if let parsed:AnyObject = SocketIOClient.parseData(asArray) {
|
||||
if ackNum == "" {
|
||||
self.handleEvent(event, data: parsed)
|
||||
} else {
|
||||
@ -573,78 +585,53 @@ class SocketIOClient {
|
||||
wantsAck: ackNum.toInt(), withAckType: 3)
|
||||
}
|
||||
return
|
||||
} else if let strData = data {
|
||||
// There are multiple items in the message
|
||||
// Turn it into a String and run it through
|
||||
// parseData to try and get an array.
|
||||
let asArray = "[\(strData)]"
|
||||
if let parsed:AnyObject = SocketIOClient.parseData(asArray) {
|
||||
if ackNum == "" {
|
||||
self.handleEvent(event, data: parsed)
|
||||
} else {
|
||||
self.currentAck = ackNum.toInt()!
|
||||
self.handleEvent(event, data: parsed, isInternalMessage: false,
|
||||
wantsAck: ackNum.toInt(), withAckType: 3)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check for no item event
|
||||
let noItemMessage = RegexMutable(messagePart)["\\[\"(.*?)\"]$"].groups()
|
||||
if noItemMessage != nil && noItemMessage.count == 2 {
|
||||
let event = noItemMessage[1]
|
||||
if ackNum == "" {
|
||||
self.handleEvent(event, data: nil)
|
||||
} else {
|
||||
self.currentAck = ackNum.toInt()!
|
||||
self.handleEvent(event, data: nil, isInternalMessage: false,
|
||||
wantsAck: ackNum.toInt(), withAckType: 3)
|
||||
}
|
||||
return
|
||||
}
|
||||
} else if messageGroups[1].hasPrefix("3") {
|
||||
let arr = Array(messageGroups[1])
|
||||
var ackNum:String
|
||||
let nsp = messageGroups[2]
|
||||
|
||||
if nsp == "" && self.nsp != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if nsp == "" {
|
||||
ackNum = String(arr[1...arr.count-1])
|
||||
}
|
||||
|
||||
// Check for no item event
|
||||
let noItemMessage = RegexMutable(messagePart)["\\[\"(.*?)\"]$"].groups()
|
||||
if noItemMessage != nil && noItemMessage.count == 2 {
|
||||
let event = noItemMessage[1]
|
||||
if ackNum == "" {
|
||||
self.handleEvent(event, data: nil)
|
||||
} else {
|
||||
ackNum = messageGroups[3]
|
||||
self.currentAck = ackNum.toInt()!
|
||||
self.handleEvent(event, data: nil, isInternalMessage: false,
|
||||
wantsAck: ackNum.toInt(), withAckType: 3)
|
||||
}
|
||||
|
||||
let ackData:AnyObject? = SocketIOClient.parseData(messageGroups[4])
|
||||
self.handleAck(ackNum.toInt()!, data: ackData)
|
||||
|
||||
return
|
||||
}
|
||||
/**
|
||||
End Check for message
|
||||
**/
|
||||
} else if messageGroups[1].hasPrefix("3") {
|
||||
let arr = Array(messageGroups[1])
|
||||
var ackNum:String
|
||||
let nsp = messageGroups[2]
|
||||
|
||||
// Check for message with binary placeholders
|
||||
self.parseBinaryMessage(message: message!)
|
||||
if nsp == "" && self.nsp != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if nsp == "" {
|
||||
ackNum = String(arr[1...arr.count-1])
|
||||
} else {
|
||||
ackNum = messageGroups[3]
|
||||
}
|
||||
|
||||
let ackData:AnyObject? = SocketIOClient.parseData(messageGroups[4])
|
||||
self.handleAck(ackNum.toInt()!, data: ackData)
|
||||
|
||||
return
|
||||
}
|
||||
/**
|
||||
End Check for message
|
||||
**/
|
||||
|
||||
// Message is binary
|
||||
if let binary = message as? NSData {
|
||||
if self.waitingData.isEmpty {
|
||||
return
|
||||
}
|
||||
|
||||
self.parseBinaryData(binary)
|
||||
}
|
||||
// Check for message with binary placeholders
|
||||
self.parseBinaryMessage(message: stringMessage)
|
||||
}
|
||||
|
||||
// Tries to parse a message that contains binary
|
||||
private func parseBinaryMessage(#message:AnyObject) {
|
||||
|
||||
// println(message)
|
||||
if let stringMessage = message as? String {
|
||||
var mutMessage = RegexMutable(stringMessage)
|
||||
@ -731,7 +718,7 @@ class SocketIOClient {
|
||||
}
|
||||
|
||||
// Handles binary data
|
||||
private func parseBinaryData(data:NSData) {
|
||||
func parseBinaryData(data:NSData) {
|
||||
let shouldExecute = self.waitingData[0].addData(data)
|
||||
|
||||
if shouldExecute {
|
||||
@ -806,8 +793,6 @@ class SocketIOClient {
|
||||
target: self!, selector: "tryReconnect", userInfo: nil, repeats: true)
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
self.handleEvent("reconnectAttempt", data: self.reconnectAttempts - self.currentReconnectAttempt,
|
||||
|
||||
714
SwiftIO/WebSocket.swift
Normal file
714
SwiftIO/WebSocket.swift
Normal file
@ -0,0 +1,714 @@
|
||||
//////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
//
|
||||
// Websocket.swift
|
||||
//
|
||||
// Created by Dalton Cherry on 7/16/14.
|
||||
//
|
||||
//////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
import Foundation
|
||||
import CoreFoundation
|
||||
|
||||
public protocol WebSocketDelegate: class {
|
||||
func websocketDidConnect(socket: WebSocket)
|
||||
func websocketDidDisconnect(socket: WebSocket, error: NSError?)
|
||||
func websocketDidReceiveMessage(socket: WebSocket, text: String)
|
||||
func websocketDidReceiveData(socket: WebSocket, data: NSData)
|
||||
}
|
||||
|
||||
public class WebSocket : NSObject, NSStreamDelegate {
|
||||
|
||||
enum OpCode : UInt8 {
|
||||
case ContinueFrame = 0x0
|
||||
case TextFrame = 0x1
|
||||
case BinaryFrame = 0x2
|
||||
//3-7 are reserved.
|
||||
case ConnectionClose = 0x8
|
||||
case Ping = 0x9
|
||||
case Pong = 0xA
|
||||
//B-F reserved.
|
||||
}
|
||||
|
||||
enum CloseCode : UInt16 {
|
||||
case Normal = 1000
|
||||
case GoingAway = 1001
|
||||
case ProtocolError = 1002
|
||||
case ProtocolUnhandledType = 1003
|
||||
// 1004 reserved.
|
||||
case NoStatusReceived = 1005
|
||||
//1006 reserved.
|
||||
case Encoding = 1007
|
||||
case PolicyViolated = 1008
|
||||
case MessageTooBig = 1009
|
||||
}
|
||||
|
||||
enum InternalErrorCode : UInt16 {
|
||||
// 0-999 WebSocket status codes not used
|
||||
case OutputStreamWriteError = 1
|
||||
}
|
||||
|
||||
//Where the callback is executed. It defaults to the main UI thread queue.
|
||||
public var queue = dispatch_get_main_queue()
|
||||
|
||||
var optionalProtocols : Array<String>?
|
||||
//Constant Values.
|
||||
let headerWSUpgradeName = "Upgrade"
|
||||
let headerWSUpgradeValue = "websocket"
|
||||
let headerWSHostName = "Host"
|
||||
let headerWSConnectionName = "Connection"
|
||||
let headerWSConnectionValue = "Upgrade"
|
||||
let headerWSProtocolName = "Sec-WebSocket-Protocol"
|
||||
let headerWSVersionName = "Sec-WebSocket-Version"
|
||||
let headerWSVersionValue = "13"
|
||||
let headerWSKeyName = "Sec-WebSocket-Key"
|
||||
let headerOriginName = "Origin"
|
||||
let headerWSAcceptName = "Sec-WebSocket-Accept"
|
||||
let BUFFER_MAX = 2048
|
||||
let FinMask: UInt8 = 0x80
|
||||
let OpCodeMask: UInt8 = 0x0F
|
||||
let RSVMask: UInt8 = 0x70
|
||||
let MaskMask: UInt8 = 0x80
|
||||
let PayloadLenMask: UInt8 = 0x7F
|
||||
let MaxFrameSize: Int = 32
|
||||
|
||||
class WSResponse {
|
||||
var isFin = false
|
||||
var code: OpCode = .ContinueFrame
|
||||
var bytesLeft = 0
|
||||
var frameCount = 0
|
||||
var buffer: NSMutableData?
|
||||
}
|
||||
|
||||
public weak var delegate: WebSocketDelegate?
|
||||
private var url: NSURL
|
||||
private var inputStream: NSInputStream?
|
||||
private var outputStream: NSOutputStream?
|
||||
private var isRunLoop = false
|
||||
private var connected = false
|
||||
private var writeQueue: NSOperationQueue?
|
||||
private var readStack = Array<WSResponse>()
|
||||
private var inputQueue = Array<NSData>()
|
||||
private var fragBuffer: NSData?
|
||||
public var headers = Dictionary<String,String>()
|
||||
public var voipEnabled = false
|
||||
public var selfSignedSSL = false
|
||||
private var connectedBlock: ((Void) -> Void)? = nil
|
||||
private var disconnectedBlock: ((NSError?) -> Void)? = nil
|
||||
private var receivedTextBlock: ((String) -> Void)? = nil
|
||||
private var receivedDataBlock: ((NSData) -> Void)? = nil
|
||||
public var isConnected :Bool {
|
||||
return connected
|
||||
}
|
||||
|
||||
//init the websocket with a url
|
||||
public init(url: NSURL) {
|
||||
self.url = url
|
||||
}
|
||||
//used for setting protocols.
|
||||
public convenience init(url: NSURL, protocols: Array<String>) {
|
||||
self.init(url: url)
|
||||
optionalProtocols = protocols
|
||||
}
|
||||
//closure based instead of the delegate
|
||||
public convenience init(url: NSURL, protocols: Array<String>, connect:((Void) -> Void), disconnect:((NSError?) -> Void), text:((String) -> Void), data:(NSData) -> Void) {
|
||||
self.init(url: url, protocols: protocols)
|
||||
connectedBlock = connect
|
||||
disconnectedBlock = disconnect
|
||||
receivedTextBlock = text
|
||||
receivedDataBlock = data
|
||||
}
|
||||
//same as above, just shorter
|
||||
public convenience init(url: NSURL, connect:((Void) -> Void), disconnect:((NSError?) -> Void), text:((String) -> Void)) {
|
||||
self.init(url: url)
|
||||
connectedBlock = connect
|
||||
disconnectedBlock = disconnect
|
||||
receivedTextBlock = text
|
||||
}
|
||||
//same as above, just shorter
|
||||
public convenience init(url: NSURL, connect:((Void) -> Void), disconnect:((NSError?) -> Void), data:((NSData) -> Void)) {
|
||||
self.init(url: url)
|
||||
connectedBlock = connect
|
||||
disconnectedBlock = disconnect
|
||||
receivedDataBlock = data
|
||||
}
|
||||
|
||||
///Connect to the websocket server on a background thread
|
||||
public func connect() {
|
||||
dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT,0), {
|
||||
self.createHTTPRequest()
|
||||
})
|
||||
}
|
||||
|
||||
///disconnect from the websocket server
|
||||
public func disconnect() {
|
||||
writeError(CloseCode.Normal.rawValue)
|
||||
}
|
||||
|
||||
///write a string to the websocket. This sends it as a text frame.
|
||||
public func writeString(str: String) {
|
||||
dequeueWrite(str.dataUsingEncoding(NSUTF8StringEncoding)!, code: .TextFrame)
|
||||
}
|
||||
|
||||
///write binary data to the websocket. This sends it as a binary frame.
|
||||
public func writeData(data: NSData) {
|
||||
dequeueWrite(data, code: .BinaryFrame)
|
||||
}
|
||||
|
||||
//write a ping to the websocket. This sends it as a control frame.
|
||||
//yodel a sound to the planet. This sends it as an astroid. http://youtu.be/Eu5ZJELRiJ8?t=42s
|
||||
public func writePing(data: NSData) {
|
||||
dequeueWrite(data, code: .Ping)
|
||||
}
|
||||
//private methods below!
|
||||
|
||||
//private method that starts the connection
|
||||
private func createHTTPRequest() {
|
||||
|
||||
let str: NSString = url.absoluteString!
|
||||
let urlRequest = CFHTTPMessageCreateRequest(kCFAllocatorDefault, "GET",
|
||||
url, kCFHTTPVersion1_1)
|
||||
|
||||
var port = url.port
|
||||
if port == nil {
|
||||
if url.scheme == "wss" || url.scheme == "https" {
|
||||
port = 443
|
||||
} else {
|
||||
port = 80
|
||||
}
|
||||
}
|
||||
self.addHeader(urlRequest, key: headerWSUpgradeName, val: headerWSUpgradeValue)
|
||||
self.addHeader(urlRequest, key: headerWSConnectionName, val: headerWSConnectionValue)
|
||||
if let protocols = optionalProtocols {
|
||||
self.addHeader(urlRequest, key: headerWSProtocolName, val: ",".join(protocols))
|
||||
}
|
||||
self.addHeader(urlRequest, key: headerWSVersionName, val: headerWSVersionValue)
|
||||
self.addHeader(urlRequest, key: headerWSKeyName, val: self.generateWebSocketKey())
|
||||
self.addHeader(urlRequest, key: headerOriginName, val: url.absoluteString!)
|
||||
self.addHeader(urlRequest, key: headerWSHostName, val: "\(url.host!):\(port!)")
|
||||
for (key,value) in headers {
|
||||
self.addHeader(urlRequest, key: key, val: value)
|
||||
}
|
||||
|
||||
let serializedRequest: NSData = CFHTTPMessageCopySerializedMessage(urlRequest.takeUnretainedValue()).takeUnretainedValue()
|
||||
self.initStreamsWithData(serializedRequest, Int(port!))
|
||||
}
|
||||
//Add a header to the CFHTTPMessage by using the NSString bridges to CFString
|
||||
private func addHeader(urlRequest: Unmanaged<CFHTTPMessage>,key: String, val: String) {
|
||||
let nsKey: NSString = key
|
||||
let nsVal: NSString = val
|
||||
CFHTTPMessageSetHeaderFieldValue(urlRequest.takeUnretainedValue(),
|
||||
nsKey,
|
||||
nsVal)
|
||||
}
|
||||
//generate a websocket key as needed in rfc
|
||||
private func generateWebSocketKey() -> String {
|
||||
var key = ""
|
||||
let seed = 16
|
||||
for (var i = 0; i < seed; i++) {
|
||||
let uni = UnicodeScalar(UInt32(97 + arc4random_uniform(25)))
|
||||
key += "\(Character(uni))"
|
||||
}
|
||||
var data = key.dataUsingEncoding(NSUTF8StringEncoding)
|
||||
var baseKey = data?.base64EncodedStringWithOptions(NSDataBase64EncodingOptions(0))
|
||||
return baseKey!
|
||||
}
|
||||
//Start the stream connection and write the data to the output stream
|
||||
private func initStreamsWithData(data: NSData, _ 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: NSString = url.host!
|
||||
CFStreamCreatePairWithSocketToHost(nil, h, UInt32(port), &readStream, &writeStream)
|
||||
inputStream = readStream!.takeUnretainedValue()
|
||||
outputStream = writeStream!.takeUnretainedValue()
|
||||
|
||||
inputStream!.delegate = self
|
||||
outputStream!.delegate = self
|
||||
if url.scheme == "wss" || url.scheme == "https" {
|
||||
inputStream!.setProperty(NSStreamSocketSecurityLevelNegotiatedSSL, forKey: NSStreamSocketSecurityLevelKey)
|
||||
outputStream!.setProperty(NSStreamSocketSecurityLevelNegotiatedSSL, forKey: NSStreamSocketSecurityLevelKey)
|
||||
}
|
||||
if self.voipEnabled {
|
||||
inputStream!.setProperty(NSStreamNetworkServiceTypeVoIP, forKey: NSStreamNetworkServiceType)
|
||||
outputStream!.setProperty(NSStreamNetworkServiceTypeVoIP, forKey: NSStreamNetworkServiceType)
|
||||
}
|
||||
if self.selfSignedSSL {
|
||||
let settings: Dictionary<NSObject, NSObject> = [kCFStreamSSLValidatesCertificateChain: NSNumber(bool:false), kCFStreamSSLPeerName: kCFNull]
|
||||
inputStream!.setProperty(settings, forKey: kCFStreamPropertySSLSettings)
|
||||
outputStream!.setProperty(settings, forKey: kCFStreamPropertySSLSettings)
|
||||
}
|
||||
isRunLoop = true
|
||||
inputStream!.scheduleInRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
|
||||
outputStream!.scheduleInRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
|
||||
inputStream!.open()
|
||||
outputStream!.open()
|
||||
let bytes = UnsafePointer<UInt8>(data.bytes)
|
||||
outputStream!.write(bytes, maxLength: data.length)
|
||||
while(isRunLoop) {
|
||||
NSRunLoop.currentRunLoop().runMode(NSDefaultRunLoopMode, beforeDate: NSDate.distantFuture() as NSDate)
|
||||
}
|
||||
}
|
||||
//delegate for the stream methods. Processes incoming bytes
|
||||
func stream(aStream: NSStream!, handleEvent eventCode: NSStreamEvent) {
|
||||
|
||||
if eventCode == .HasBytesAvailable {
|
||||
if(aStream == inputStream) {
|
||||
processInputStream()
|
||||
}
|
||||
} else if eventCode == .ErrorOccurred {
|
||||
disconnectStream(aStream!.streamError)
|
||||
} else if eventCode == .EndEncountered {
|
||||
disconnectStream(nil)
|
||||
}
|
||||
}
|
||||
//disconnect the stream object
|
||||
private func disconnectStream(error: NSError?) {
|
||||
if writeQueue != nil {
|
||||
writeQueue!.waitUntilAllOperationsAreFinished()
|
||||
}
|
||||
inputStream!.removeFromRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
|
||||
outputStream!.removeFromRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
|
||||
inputStream!.close()
|
||||
outputStream!.close()
|
||||
inputStream = nil
|
||||
outputStream = nil
|
||||
isRunLoop = false
|
||||
connected = false
|
||||
dispatch_async(queue,{
|
||||
if let disconnectBlock = self.disconnectedBlock {
|
||||
disconnectBlock(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
})
|
||||
}
|
||||
|
||||
///handles the incoming bytes and sending them to the proper processing method
|
||||
private func processInputStream() {
|
||||
let buf = NSMutableData(capacity: BUFFER_MAX)
|
||||
var buffer = UnsafeMutablePointer<UInt8>(buf!.bytes)
|
||||
let length = inputStream!.read(buffer, maxLength: BUFFER_MAX)
|
||||
if length > 0 {
|
||||
if !connected {
|
||||
connected = processHTTP(buffer, bufferLen: length)
|
||||
if !connected {
|
||||
dispatch_async(queue,{
|
||||
//self.workaroundMethod()
|
||||
let error = self.errorWithDetail("Invalid HTTP upgrade", code: 1)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
})
|
||||
}
|
||||
} else {
|
||||
var process = false
|
||||
if inputQueue.count == 0 {
|
||||
process = true
|
||||
}
|
||||
inputQueue.append(NSData(bytes: buffer, length: length))
|
||||
if process {
|
||||
dequeueInput()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
///dequeue the incoming input so it is processed in order
|
||||
private func dequeueInput() {
|
||||
if inputQueue.count > 0 {
|
||||
let data = inputQueue[0]
|
||||
var work = data
|
||||
if (fragBuffer != nil) {
|
||||
var combine = NSMutableData(data: fragBuffer!)
|
||||
combine.appendData(data)
|
||||
work = combine
|
||||
fragBuffer = nil
|
||||
}
|
||||
let buffer = UnsafePointer<UInt8>(work.bytes)
|
||||
processRawMessage(buffer, bufferLen: work.length)
|
||||
inputQueue = inputQueue.filter{$0 != data}
|
||||
dequeueInput()
|
||||
}
|
||||
}
|
||||
///Finds the HTTP Packet in the TCP stream, by looking for the CRLF.
|
||||
private func processHTTP(buffer: UnsafePointer<UInt8>, bufferLen: Int) -> Bool {
|
||||
let CRLFBytes = [UInt8("\r"), UInt8("\n"), UInt8("\r"), UInt8("\n")]
|
||||
var k = 0
|
||||
var totalSize = 0
|
||||
for var i = 0; i < bufferLen; i++ {
|
||||
if buffer[i] == CRLFBytes[k] {
|
||||
k++
|
||||
if k == 3 {
|
||||
totalSize = i + 1
|
||||
break
|
||||
}
|
||||
} else {
|
||||
k = 0
|
||||
}
|
||||
}
|
||||
if totalSize > 0 {
|
||||
if validateResponse(buffer, bufferLen: totalSize) {
|
||||
dispatch_async(queue,{
|
||||
//self.workaroundMethod()
|
||||
if let connectBlock = self.connectedBlock {
|
||||
connectBlock()
|
||||
}
|
||||
self.delegate?.websocketDidConnect(self)
|
||||
})
|
||||
totalSize += 1 //skip the last \n
|
||||
let restSize = bufferLen - totalSize
|
||||
if restSize > 0 {
|
||||
processRawMessage((buffer+totalSize),bufferLen: restSize)
|
||||
}
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
///validates the HTTP is a 101 as per the RFC spec
|
||||
private func validateResponse(buffer: UnsafePointer<UInt8>, bufferLen: Int) -> Bool {
|
||||
let response = CFHTTPMessageCreateEmpty(kCFAllocatorDefault, 0)
|
||||
CFHTTPMessageAppendBytes(response.takeUnretainedValue(), buffer, bufferLen)
|
||||
if CFHTTPMessageGetResponseStatusCode(response.takeUnretainedValue()) != 101 {
|
||||
return false
|
||||
}
|
||||
let cfHeaders = CFHTTPMessageCopyAllHeaderFields(response.takeUnretainedValue())
|
||||
let headers: NSDictionary = cfHeaders.takeUnretainedValue()
|
||||
let acceptKey = headers[headerWSAcceptName] as NSString
|
||||
if acceptKey.length > 0 {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
///process the websocket data
|
||||
private func processRawMessage(buffer: UnsafePointer<UInt8>, bufferLen: Int) {
|
||||
var response = readStack.last
|
||||
if response != nil && bufferLen < 2 {
|
||||
fragBuffer = NSData(bytes: buffer, length: bufferLen)
|
||||
return
|
||||
}
|
||||
if response != nil && response!.bytesLeft > 0 {
|
||||
let resp = response!
|
||||
var len = resp.bytesLeft
|
||||
var extra = bufferLen - resp.bytesLeft
|
||||
if resp.bytesLeft > bufferLen {
|
||||
len = bufferLen
|
||||
extra = 0
|
||||
}
|
||||
resp.bytesLeft -= len
|
||||
resp.buffer?.appendData(NSData(bytes: buffer, length: len))
|
||||
processResponse(resp)
|
||||
var offset = bufferLen - extra
|
||||
if extra > 0 {
|
||||
processExtra((buffer+offset), bufferLen: extra)
|
||||
}
|
||||
return
|
||||
} else {
|
||||
let isFin = (FinMask & buffer[0])
|
||||
let receivedOpcode = (OpCodeMask & buffer[0])
|
||||
let isMasked = (MaskMask & buffer[1])
|
||||
let payloadLen = (PayloadLenMask & buffer[1])
|
||||
var offset = 2
|
||||
if((isMasked > 0 || (RSVMask & buffer[0]) > 0) && receivedOpcode != OpCode.Pong.rawValue) {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("masked and rsv data is not currently supported", code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
let isControlFrame = (receivedOpcode == OpCode.ConnectionClose.rawValue || receivedOpcode == OpCode.Ping.rawValue)
|
||||
if !isControlFrame && (receivedOpcode != OpCode.BinaryFrame.rawValue && receivedOpcode != OpCode.ContinueFrame.rawValue &&
|
||||
receivedOpcode != OpCode.TextFrame.rawValue && receivedOpcode != OpCode.Pong.rawValue) {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("unknown opcode: \(receivedOpcode)", code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
if isControlFrame && isFin == 0 {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("control frames can't be fragmented", code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
if receivedOpcode == OpCode.ConnectionClose.rawValue {
|
||||
var code = CloseCode.Normal.rawValue
|
||||
if payloadLen == 1 {
|
||||
code = CloseCode.ProtocolError.rawValue
|
||||
} else if payloadLen > 1 {
|
||||
var codeBuffer = UnsafePointer<UInt16>((buffer+offset))
|
||||
code = codeBuffer[0].byteSwapped
|
||||
if code < 1000 || (code > 1003 && code < 1007) || (code > 1011 && code < 3000) {
|
||||
code = CloseCode.ProtocolError.rawValue
|
||||
}
|
||||
offset += 2
|
||||
}
|
||||
if payloadLen > 2 {
|
||||
let len = Int(payloadLen-2)
|
||||
if len > 0 {
|
||||
let bytes = UnsafePointer<UInt8>((buffer+offset))
|
||||
var str: NSString? = NSString(data: NSData(bytes: bytes, length: len), encoding: NSUTF8StringEncoding)
|
||||
if str == nil {
|
||||
code = CloseCode.ProtocolError.rawValue
|
||||
}
|
||||
}
|
||||
}
|
||||
let error = self.errorWithDetail("connection closed by server", code: code)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(code)
|
||||
return
|
||||
}
|
||||
if isControlFrame && payloadLen > 125 {
|
||||
writeError(CloseCode.ProtocolError.rawValue)
|
||||
return
|
||||
}
|
||||
var dataLength = UInt64(payloadLen)
|
||||
if dataLength == 127 {
|
||||
let bytes = UnsafePointer<UInt64>((buffer+offset))
|
||||
dataLength = bytes[0].byteSwapped
|
||||
offset += sizeof(UInt64)
|
||||
} else if dataLength == 126 {
|
||||
let bytes = UnsafePointer<UInt16>((buffer+offset))
|
||||
dataLength = UInt64(bytes[0].byteSwapped)
|
||||
offset += sizeof(UInt16)
|
||||
}
|
||||
var len = dataLength
|
||||
if dataLength > UInt64(bufferLen) {
|
||||
len = UInt64(bufferLen-offset)
|
||||
}
|
||||
var data: NSData!
|
||||
if len < 0 {
|
||||
len = 0
|
||||
data = NSData()
|
||||
} else {
|
||||
data = NSData(bytes: UnsafePointer<UInt8>((buffer+offset)), length: Int(len))
|
||||
}
|
||||
if receivedOpcode == OpCode.Pong.rawValue {
|
||||
let step = Int(offset+len)
|
||||
let extra = bufferLen-step
|
||||
if extra > 0 {
|
||||
processRawMessage((buffer+step), bufferLen: extra)
|
||||
}
|
||||
return
|
||||
}
|
||||
var response = readStack.last
|
||||
if isControlFrame {
|
||||
response = nil //don't append pings
|
||||
}
|
||||
if isFin == 0 && receivedOpcode == OpCode.ContinueFrame.rawValue && response == nil {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("continue frame before a binary or text frame", code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
var isNew = false
|
||||
if(response == nil) {
|
||||
if receivedOpcode == OpCode.ContinueFrame.rawValue {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("first frame can't be a continue frame",
|
||||
code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
isNew = true
|
||||
response = WSResponse()
|
||||
response!.code = OpCode(rawValue: receivedOpcode)!
|
||||
response!.bytesLeft = Int(dataLength)
|
||||
response!.buffer = NSMutableData(data: data)
|
||||
} else {
|
||||
if receivedOpcode == OpCode.ContinueFrame.rawValue {
|
||||
response!.bytesLeft = Int(dataLength)
|
||||
} else {
|
||||
let errCode = CloseCode.ProtocolError.rawValue
|
||||
let error = self.errorWithDetail("second and beyond of fragment message must be a continue frame",
|
||||
code: errCode)
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
writeError(errCode)
|
||||
return
|
||||
}
|
||||
response!.buffer!.appendData(data)
|
||||
}
|
||||
if response != nil {
|
||||
response!.bytesLeft -= Int(len)
|
||||
response!.frameCount++
|
||||
response!.isFin = isFin > 0 ? true : false
|
||||
if(isNew) {
|
||||
readStack.append(response!)
|
||||
}
|
||||
processResponse(response!)
|
||||
}
|
||||
|
||||
let step = Int(offset+len)
|
||||
let extra = bufferLen-step
|
||||
if(extra > 0) {
|
||||
processExtra((buffer+step), bufferLen: extra)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
///process the extra of a buffer
|
||||
private func processExtra(buffer: UnsafePointer<UInt8>, bufferLen: Int) {
|
||||
if bufferLen < 2 {
|
||||
fragBuffer = NSData(bytes: buffer, length: bufferLen)
|
||||
} else {
|
||||
processRawMessage(buffer, bufferLen: bufferLen)
|
||||
}
|
||||
}
|
||||
|
||||
///process the finished response of a buffer
|
||||
private func processResponse(response: WSResponse) -> Bool {
|
||||
if response.isFin && response.bytesLeft <= 0 {
|
||||
if response.code == .Ping {
|
||||
let data = response.buffer! //local copy so it is perverse for writing
|
||||
dequeueWrite(data, code: OpCode.Pong)
|
||||
} else if response.code == .TextFrame {
|
||||
var str: NSString? = NSString(data: response.buffer!, encoding: NSUTF8StringEncoding)
|
||||
if str == nil {
|
||||
writeError(CloseCode.Encoding.rawValue)
|
||||
return false
|
||||
}
|
||||
dispatch_async(queue,{
|
||||
if let textBlock = self.receivedTextBlock{
|
||||
textBlock(str!)
|
||||
}
|
||||
self.delegate?.websocketDidReceiveMessage(self, text: str!)
|
||||
})
|
||||
} else if response.code == .BinaryFrame {
|
||||
let data = response.buffer! //local copy so it is perverse for writing
|
||||
dispatch_async(queue,{
|
||||
//self.workaroundMethod()
|
||||
if let dataBlock = self.receivedDataBlock{
|
||||
dataBlock(data)
|
||||
}
|
||||
self.delegate?.websocketDidReceiveData(self, data: data)
|
||||
})
|
||||
}
|
||||
readStack.removeLast()
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
///Create an error
|
||||
private func errorWithDetail(detail: String, code: UInt16) -> NSError {
|
||||
var details = Dictionary<String,String>()
|
||||
details[NSLocalizedDescriptionKey] = detail
|
||||
return NSError(domain: "Websocket", code: Int(code), userInfo: details)
|
||||
}
|
||||
|
||||
///write a an error to the socket
|
||||
private func writeError(code: UInt16) {
|
||||
let buf = NSMutableData(capacity: sizeof(UInt16))
|
||||
var buffer = UnsafeMutablePointer<UInt16>(buf!.bytes)
|
||||
buffer[0] = code.byteSwapped
|
||||
dequeueWrite(NSData(bytes: buffer, length: sizeof(UInt16)), code: .ConnectionClose)
|
||||
}
|
||||
///used to write things to the stream in a
|
||||
private func dequeueWrite(data: NSData, code: OpCode) {
|
||||
if writeQueue == nil {
|
||||
writeQueue = NSOperationQueue()
|
||||
writeQueue!.maxConcurrentOperationCount = 1
|
||||
}
|
||||
writeQueue!.addOperationWithBlock {
|
||||
//stream isn't ready, let's wait
|
||||
var tries = 0;
|
||||
while self.outputStream == nil || !self.connected {
|
||||
if(tries < 5) {
|
||||
sleep(1);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
tries++;
|
||||
}
|
||||
if !self.connected {
|
||||
return
|
||||
}
|
||||
var offset = 2
|
||||
UINT16_MAX
|
||||
let bytes = UnsafeMutablePointer<UInt8>(data.bytes)
|
||||
let dataLength = data.length
|
||||
let frame = NSMutableData(capacity: dataLength + self.MaxFrameSize)
|
||||
let buffer = UnsafeMutablePointer<UInt8>(frame!.mutableBytes)
|
||||
buffer[0] = self.FinMask | code.rawValue
|
||||
if dataLength < 126 {
|
||||
buffer[1] = CUnsignedChar(dataLength)
|
||||
} else if dataLength <= Int(UInt16.max) {
|
||||
buffer[1] = 126
|
||||
var sizeBuffer = UnsafeMutablePointer<UInt16>((buffer+offset))
|
||||
sizeBuffer[0] = UInt16(dataLength).byteSwapped
|
||||
offset += sizeof(UInt16)
|
||||
} else {
|
||||
buffer[1] = 127
|
||||
var sizeBuffer = UnsafeMutablePointer<UInt64>((buffer+offset))
|
||||
sizeBuffer[0] = UInt64(dataLength).byteSwapped
|
||||
offset += sizeof(UInt64)
|
||||
}
|
||||
buffer[1] |= self.MaskMask
|
||||
var maskKey = UnsafeMutablePointer<UInt8>(buffer + offset)
|
||||
SecRandomCopyBytes(kSecRandomDefault, UInt(sizeof(UInt32)), maskKey)
|
||||
offset += sizeof(UInt32)
|
||||
|
||||
for (var i = 0; i < dataLength; i++) {
|
||||
buffer[offset] = bytes[i] ^ maskKey[i % sizeof(UInt32)]
|
||||
offset += 1
|
||||
}
|
||||
var total = 0
|
||||
while true {
|
||||
if self.outputStream == nil {
|
||||
break
|
||||
}
|
||||
let writeBuffer = UnsafePointer<UInt8>(frame!.bytes+total)
|
||||
var len = self.outputStream?.write(writeBuffer, maxLength: offset-total)
|
||||
if len == nil || len! < 0 {
|
||||
var error: NSError?
|
||||
if let streamError = self.outputStream?.streamError {
|
||||
error = streamError
|
||||
} else {
|
||||
let errCode = InternalErrorCode.OutputStreamWriteError.rawValue
|
||||
error = self.errorWithDetail("output stream error during write", code: errCode)
|
||||
}
|
||||
if let disconnect = self.disconnectedBlock {
|
||||
disconnect(error)
|
||||
}
|
||||
self.delegate?.websocketDidDisconnect(self, error: error)
|
||||
break
|
||||
} else {
|
||||
total += len!
|
||||
}
|
||||
if total >= offset {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user