Use POSIX functions to build Input/Output streams

Author: rhx

Sources/NetServiceListener.swift

@@ -45,13 +45,8 @@ extension DNSSDNetService {
             let this = Unmanaged<DNSSDNetService>.fromOpaque(info).takeRetainedValue()
             guard let delegate = this.delegate else { return }
             guard let socket = data?.assumingMemoryBound(to: CFSocketNativeHandle.self).pointee else { return }
-            var readStream: Unmanaged<CFReadStream>?
-            var writeStream: Unmanaged<CFWriteStream>?
-            CFStreamCreatePairWithSocket(kCFAllocatorDefault, socket, &readStream, &writeStream)
-            guard let r = readStream?.takeRetainedValue(),
-                  let w = writeStream?.takeRetainedValue() else { return }
-            let inputStream = DNSSDNetServiceInputStream(r)
-            let outputStream = DNSSDNetServiceOutputStream(w)
+            let inputStream = DNSSDNetServiceInputStream(socket)
+            let outputStream = DNSSDNetServiceOutputStream(socket)
             delegate.netService(this, didAcceptConnectionWith: inputStream, outputStream: outputStream)
         }
         let this = Unmanaged.passUnretained(self).toOpaque()

Sources/NetServiceStreams.swift

@@ -5,8 +5,6 @@
 //  Created by Rene Hexel on 10/12/16.
 //  Copyright © 2016 René Hexel.  All rights reserved.
 //
-//#if os(Linux)
-
 import CoreFoundation
 import Foundation
 
@@ -15,50 +13,98 @@ import Foundation
     private func CFStreamPropertyKey(rawValue: CFString) -> CFString {
         return rawValue
     }
+
+    /// Set bind the given socket to the given address and listen for connections
+    ///
+    /// - Parameters:
+    ///   - s: `CFSocket` to bind
+    ///   - address: IP address to bind to (Data containing a `struct sockaddr*`)
+    /// - Returns: `0` if successful, an error code otherwise
+    @discardableResult
+    func CFSocketSetAddress(_ s: CFSocket, _ address: CFData!) -> CFSocketError {
+        let len = socklen_t(CFDataGetLength(address))
+        guard address != nil,
+              len >= socklen_t(MemoryLayout<sockaddr_in>.size),
+              CFSocketIsValid(s) else {
+                return CFSocketError(kCFSocketError)
+        }
+        let sock = CFSocketGetNative(s)
+        return CFDataGetBytePtr(address).withMemoryRebound(to: sockaddr.self, capacity: 1) { (addr: UnsafePointer<sockaddr>!) -> CFSocketError in
+            guard addr != nil else { return CFSocketError(kCFSocketError) }
+            guard bind(sock, addr, len) == 0,
+                  listen(sock, 256) == 0 else { return CFSocketError(errno) }
+            return CFSocketError(kCFSocketSuccess)
+        }
+    }
 #else
     private let utf8 = CFStringBuiltInEncodings.UTF8.rawValue
 #endif
 
+extension CFSocketNativeHandle {
+    func get(_ buffer: UnsafeMutableRawPointer, maxLength len: Int) -> Int {
+        return Int(read(Int32(self), buffer, ssize_t(len)))
+    }
+
+    func put(_ buffer: UnsafeRawPointer, maxLength len: Int) -> Int {
+        return Int(write(Int32(self), buffer, ssize_t(len)))
+    }
+
+    func closeReadingEnd() {
+        shutdown(Int32(self), Int32(SHUT_RD))
+    }
+
+    func closeWritingEnd() {
+        shutdown(Int32(self), Int32(SHUT_WR))
+    }
+
+    func has(events: Int16 = Int16(POLLIN)) -> Bool {
+        var fd = pollfd(fd: Int32(self), events: events, revents: 0)
+        return poll(&fd, 1, 0) > 0
+    }
+}
+
 public class DNSSDNetServiceInputStream: InputStream {
-    var cfStream: CFReadStream
+    var sock: CFSocketNativeHandle
+    var status = Status.open
 
-    /// Initialise from a `CFReadStream`
+    /// Initialise from a `CFSocketNativeHandle`
     ///
-    /// - Parameter stream: CoreFoundation read stream to initialise from
-    public init(_ stream: CFReadStream) {
-        cfStream = stream
+    /// - Parameter socket: Already connected socket to initialise from
+    public init(_ socket: CFSocketNativeHandle) {
+        sock = socket
         super.init(data: Data())
     }
 
     public override func read(_ buffer: UnsafeMutablePointer<UInt8>, maxLength len: Int) -> Int {
-        return CFReadStreamRead(cfStream, buffer, CFIndex(len))
+        return sock.get(buffer, maxLength: len)
     }
 
     public override func getBuffer(_ buffer: UnsafeMutablePointer<UnsafeMutablePointer<UInt8>?>, length len: UnsafeMutablePointer<Int>) -> Bool {
         return false
     }
 
     public override var hasBytesAvailable: Bool {
-        return CFReadStreamHasBytesAvailable(cfStream)
+        return sock.has(events: Int16(POLLIN))
     }
 
     public override func open() {
-        CFReadStreamOpen(cfStream)
     }
 
     public override func close() {
-        CFReadStreamClose(cfStream)
+        sock.closeReadingEnd()
+        status = Status.closed
     }
 
     public override var streamStatus: Status {
-        let status = CFReadStreamGetStatus(cfStream)
-        return Stream.Status(rawValue: unsafeBitCast(status, to: UInt.self))!
+        return status
     }
 }
 
 
 public class DNSSDNetServiceOutputStream: OutputStream {
-    var cfStream: CFWriteStream!
+    var sock: CFSocketNativeHandle
+    var status = Status.open
+    var properties = Dictionary<PropertyKey, PropertyValue>()
 
     #if os(Linux)
         public typealias PropertyValue = AnyObject
@@ -70,50 +116,41 @@ public class DNSSDNetServiceOutputStream: OutputStream {
         public typealias PropertyValue = Any
     #endif
 
-    public init(_ stream: CFWriteStream) {
-        cfStream = stream
+    public init(_ socket: CFSocketNativeHandle) {
+        sock = socket
         super.init(toMemory: ())
     }
 
 
     // writes the bytes from the specified buffer to the stream up to len bytes. Returns the number of bytes actually written.
     public override func write(_ buffer: UnsafePointer<UInt8>, maxLength len: Int) -> Int {
-        return  CFWriteStreamWrite(cfStream, buffer, len)
+        return sock.put(buffer, maxLength: len)
     }
 
     // returns YES if the stream can be written to or if it is impossible to tell without actually doing the write.
     public override var hasSpaceAvailable: Bool {
-        return CFWriteStreamCanAcceptBytes(cfStream)
+        return sock.has(events: Int16(POLLOUT))
     }
 
     public override func open() {
-        CFWriteStreamOpen(cfStream)
     }
 
     public override func close() {
-        CFWriteStreamClose(cfStream)
+        sock.closeReadingEnd()
+        status = Status.closed
     }
 
     public override var streamStatus: Status {
-        let status = CFWriteStreamGetStatus(cfStream)
-        return Stream.Status(rawValue: unsafeBitCast(status, to: UInt.self))!
+        return status
     }
 
     public override func property(forKey key: PropertyKey) -> PropertyValue? {
-        return key.rawValue.withCString {
-            guard let k = CFStringCreateWithCString(kCFAllocatorDefault, $0, utf8) else {
-                return nil
-            }
-            return CFWriteStreamCopyProperty(cfStream, CFStreamPropertyKey(rawValue: k))
-        }
+        return properties[key]
     }
 
     public override func setProperty(_ property: PropertyValue?, forKey key: PropertyKey) -> Bool {
-        return key.rawValue.withCString {
-            guard let k = CFStringCreateWithCString(kCFAllocatorDefault, $0, utf8) else {
-                return false
-            }
-            return CFWriteStreamSetProperty(cfStream, CFStreamPropertyKey(rawValue: k), property as AnyObject?)
-        }
+        let rv = super.setProperty(property, forKey: key)
+        properties[key] = property
+        return rv
     }
 }