Add SocketManager.listen()

Author: colemancda

Sources/Socket/Socket.swift

@@ -140,8 +140,8 @@ public struct Socket {
     }
 
     /// Listen for connections on a socket.
-    public func listen(backlog: Int = Self.maxSocketBacklog) throws {
-        try fileDescriptor.listen(backlog: backlog)
+    public func listen(backlog: Int = Self.maxSocketBacklog) async throws {
+        try await manager.listen(backlog: backlog, for: fileDescriptor)
     }
 
     /// Accept new socket.
@@ -208,4 +208,3 @@ public extension Socket.Event {
     /// Socket Event Stream
     typealias Stream = AsyncStream<Socket.Event>
 }
-

Sources/Socket/SocketManager.swift

@@ -71,6 +71,12 @@ public protocol SocketManager: AnyObject {
         to address: Address,
         for fileDescriptor: SocketDescriptor
     ) async throws
+    
+    /// Listen for incoming connections
+    func listen(
+        backlog: Int,
+        for fileDescriptor: SocketDescriptor
+    ) async throws
 }
 
 /// Socket Manager Configuration

Sources/Socket/SocketManager/AsyncSocketManager.swift

@@ -165,27 +165,24 @@ internal actor AsyncSocketManager: SocketManager {
         return try await socket.receiveMessage(length, fromAddressOf: addressType)
     }
 
+    nonisolated func listen(backlog: Int, for fileDescriptor: SocketDescriptor) async throws {
+        let socket = try await self.socket(for: fileDescriptor)
+        try await socket.listen(backlog: backlog)
+    }
+    
     /// Accept a connection on a socket.
     nonisolated func accept(for fileDescriptor: SocketDescriptor) async throws -> SocketDescriptor {
-        let socket = try await socket(for: fileDescriptor)
-        let result = try await retry(sleep: state.configuration.monitorInterval) {
-            fileDescriptor._accept(retryOnInterrupt: true)
-        }.get()
-        socket.continuation.yield(.connection)
-        return result
+        let socket = try await wait(for: .read, fileDescriptor: fileDescriptor)
+        return try await socket.accept()
     }
 
     /// Accept a connection on a socket.
     nonisolated func accept<Address: SocketAddress>(
         _ address: Address.Type,
         for fileDescriptor: SocketDescriptor
     ) async throws -> (fileDescriptor: SocketDescriptor, address: Address) {
-        let socket = try await socket(for: fileDescriptor)
-        let result = try await retry(sleep: state.configuration.monitorInterval) {
-            fileDescriptor._accept(address, retryOnInterrupt: true)
-        }.get()
-        socket.continuation.yield(.connection)
-        return result
+        let socket = try await wait(for: .read, fileDescriptor: fileDescriptor)
+        return try await socket.accept(address)
     }
 
     /// Initiate a connection on a socket.
@@ -224,7 +221,7 @@ private extension AsyncSocketManager {
         var tasks = [Task<Void, Never>]()
         while self.state.isMonitoring {
             do {
-                tasks.reserveCapacity(state.sockets.count * 2)
+                tasks.reserveCapacity(state.sockets.count)
                 // poll
                 let hasEvents = try poll(&tasks)
                 // stop monitoring if no sockets
@@ -331,39 +328,28 @@ private extension AsyncSocketManager {
     }
 
     func process(_ poll: SocketDescriptor.Poll, socket: AsyncSocketManager.SocketState, tasks: inout [Task<Void, Never>]) {
-        /*
-        let isListening = self.sockets[poll.socket]?.isListening ?? false
-        if isListening, poll.returnedEvents.contains([.read, .write]) {
-            event([.read, .write], notification: .connection, for: poll.socket)
-        } else {
+        let task = Task {
             if poll.returnedEvents.contains(.read) {
-                event(.read, notification: .read, for: poll.socket)
+                if await socket.isListening {
+                    await socket.event(.read, notification: .connection)
+                } else {
+                    await socket.event(.read, notification: .read)
+                }
             }
             if poll.returnedEvents.contains(.write) {
-                event(.write, notification: .write, for: poll.socket)
+                await socket.event(.write, notification: .write)
             }
-        }*/
-        if poll.returnedEvents.contains(.read) {
-            let task = Task(priority: state.configuration.monitorPriority) {
-                await socket.event(.read, notification: .read)
+            if poll.returnedEvents.contains(.invalidRequest) {
+                error(.badFileDescriptor, for: poll.socket)
             }
-            tasks.append(task)
-        }
-        if poll.returnedEvents.contains(.write) {
-            let task = Task(priority: state.configuration.monitorPriority) {
-                await socket.event(.write, notification: .write)
+            if poll.returnedEvents.contains(.error) {
+                error(.connectionReset, for: poll.socket)
+            }
+            if poll.returnedEvents.contains(.hangup) {
+                hangup(poll.socket)
             }
-            tasks.append(task)
-        }
-        if poll.returnedEvents.contains(.invalidRequest) {
-            error(.badFileDescriptor, for: poll.socket)
-        }
-        if poll.returnedEvents.contains(.error) {
-            error(.connectionReset, for: poll.socket)
-        }
-        if poll.returnedEvents.contains(.hangup) {
-            hangup(poll.socket)
         }
+        tasks.append(task)
     }
 
     func error(_ error: Errno, for fileDescriptor: SocketDescriptor) {
@@ -443,6 +429,19 @@ extension AsyncSocketManager.SocketState {
         didRead(bytesRead)
         return (data, address)
     }
+    
+    func listen(backlog: Int) throws {
+        try fileDescriptor.listen(backlog: backlog)
+        isListening = true
+    }
+    
+    func accept() throws -> SocketDescriptor {
+        try fileDescriptor.accept()
+    }
+    
+    func accept<Address: SocketAddress>(_ address: Address.Type) throws -> (SocketDescriptor, Address) {
+        try fileDescriptor.accept(address)
+    }
 }
 
 fileprivate extension AsyncSocketManager.SocketState {

Tests/SocketTests/SocketTests.swift

@@ -49,7 +49,7 @@ final class SocketTests: XCTestCase {
         let newConnectionTask = Task {
             XCTAssertEqual(try server.fileDescriptor.address(IPv4SocketAddress.self), address)
             NSLog("Server: Created server socket \(server.fileDescriptor)")
-            try server.listen()
+            try await server.listen()
 
             NSLog("Server: Waiting on incoming connection")
             let newConnection = try await server.accept()
@@ -103,8 +103,8 @@ final class SocketTests: XCTestCase {
         try await Task.sleep(nanoseconds: 1_000_000_000)
         await client.close()
         let clientEvents = try await clientEventsTask.value
-        XCTAssertEqual(clientEvents.count, 4)
-        XCTAssertEqual("\(clientEvents)", "[Socket.Socket.Event.write, Socket.Socket.Event.read, Socket.Socket.Event.didRead(41), Socket.Socket.Event.close]")
+        XCTAssertEqual(clientEvents.count, 5)
+        XCTAssertEqual("\(clientEvents)", "[Socket.Socket.Event.write, Socket.Socket.Event.read, Socket.Socket.Event.didRead(41), Socket.Socket.Event.read, Socket.Socket.Event.close]")
         await server.close()
         let serverEvents = try await serverEventsTask.value
         XCTAssertEqual(serverEvents.count, 2)