Swift Socket Server

gistfile1.txt

import Foundation

public class Client {
    public var socket:Socket
    public var sentdata:dispatch_data_t?
//    public var channel:dispatch_io_t
    public var hashValue: Int { return Int(socket.descriptor) }

    init(socket sock:Socket, sentdata sdata:dispatch_data_t?) {
        socket = sock
        sentdata = sdata
    }
}


public func ==(lhs:Client, rhs:Client) -> Bool {
    return lhs.hashValue == rhs.hashValue
}

public class Server {
    public let address:String
    public let port:SocketPort
    
    
    private var listener    = Socket(descriptor: -1)
    private var clients     = Array<Client>()
    private let semaphore   = dispatch_semaphore_create(1)
    private let acceptQueue = dispatch_queue_create("com.pilot.connection.accept.queue", DISPATCH_QUEUE_CONCURRENT)
    private let acceptGroup = dispatch_group_create()
    private let handleQueue = dispatch_queue_create("com.pilot.connection.handle.queue", nil)
    private let handleGroup = dispatch_group_create()
    private var handler:((AnyObject?, Client) -> Void)?

    public init(address:String = "0.0.0.0", port: SocketPort) throws {
        self.address    = address
        self.port       = port
        
        listener                = try Socket()
        listener.closeOnExec    = true
        
        try listener.bind(address, port: port)
        try listener.listen(1000)
    }
    
    private var source:dispatch_source_t?
    private var writeSource:dispatch_source_t?
    private let sourceQueue = dispatch_queue_create("com.pilot.connection.source.queue", DISPATCH_QUEUE_CONCURRENT)
    
    public func serve(handler: (AnyObject?, Client) -> Void) throws {
        guard listener.descriptor > -1 else {
            throw SocketError()
        }
        
        self.handler    = handler



        source = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, UInt(listener.descriptor), 0, sourceQueue)
        
        if let source = source {
            dispatch_source_set_event_handler(source) { [weak self] in
                guard let _self = self where _self.listener.descriptor > 0 else { fatalError("No Self") }
                
//                print("READ SOURCE TRIGGERED")
                if let incoming = try? _self.listener.accept() {
//                    print("CLIENT COUNT= \(_self.clients.count)")
                    _self.read(incoming)
                }
            }
            
            dispatch_resume(source)
        }
        
        dispatch_main()
    }

    public func stop() {
        for client in clients {
            client.socket.close()
        }

        listener.close()
    }
    
    private let _isolation = dispatch_queue_create("channel.read.queue", DISPATCH_QUEUE_CONCURRENT)
    private func read(connection:Socket) {
//        print("READING SOCKET")


        let client = Client(socket: connection, sentdata: nil)
        
        let _channel = dispatch_io_create(DISPATCH_IO_STREAM, connection.descriptor, _isolation){ [weak self] error in
            //            if error > 0 {
            //                print("Channel error: \(POSIXError(rawValue:error))")
            //                return
            //            }
            guard let _self = self else {
                return
            }

            if(error > 0) {
                _self.remove(client)
                
            }else{
                _self.remove(client)
            }
            SocketFunctions.Shutdown(client.socket.descriptor, Int32(SHUT_RDWR))
            SocketFunctions.Close(client.socket.descriptor)
            client.socket.descriptor = -1
        };
        
        connection.closeHandler = { [weak self] in
            
//            print("CLOSE HANDLER")
            guard let _self = self else {
                return
            }
//            print("CLOSING CHANNEL \(connection.descriptor)  and client socket \(client.socket.descriptor)")
            dispatch_io_close(_channel, DISPATCH_IO_STOP)

            
        }

        sync {
            self.clients.append(client)
//            self.clients.insert(client)
        }


            
//            dispatch_io_set_high_water(_channel, 8 * 1024);
            dispatch_io_set_low_water(_channel, 1);
            dispatch_io_set_interval(_channel, NSEC_PER_MSEC * 10, DISPATCH_IO_STRICT_INTERVAL);
        
            dispatch_io_read(_channel, 0, 2000, _isolation){ [weak self] done, data, error in
//                print("READING CHANNEL")
                guard let _self = self else {
                    return
                }
                
                if done {
//                    print("THIS IS DONE")
//                    _self.remove(client)
                    return
                }
                
                if error == 0 {
                    let len = dispatch_data_get_size(data)
                    if len > 0 {
                        if (client.sentdata == nil) {
                            client.sentdata = data;
                        } else {
                            client.sentdata = dispatch_data_create_concat(client.sentdata!, data);
                        }
                        
                        if len < 2000 {
                            if let d = client.sentdata as? NSData {
                                let str =  NSString(bytes: d.bytes, length: d.length, encoding: NSUTF8StringEncoding)
                                _self.handler?(str, client)
                            }
                        }
                        
                        
                    }
                }
                

                if error > 0 {
                    print("Channel error: \(error) and \(POSIXError(rawValue:error))")
                    client.socket.close()
                    return
                }
                

            };

    }
    
    private func remove(client:Client) {

//                    client.socket.close()
            sync {
                let ind = self.clients.indexOf({ (cl) -> Bool in
                    return cl == client
                })
                
                if (ind != nil) {
                    self.clients.removeAtIndex(ind!)
                }
        }

        
    }
    
    private func sync(block:Void -> Void) {
        dispatch_semaphore_wait(self.semaphore, DISPATCH_TIME_FOREVER)
        block()
        dispatch_semaphore_signal(self.semaphore)
    }
}