Simplify implementation of Dispatch-async bridge

Stress testing this approach over use of a long-lived DispatchIO channel appears to resolve the nondeterministic failures with the file descriptor being destroyed, and is easier to reason about.

Closes #21

Author: jakepetroules

Sources/SWBUtil/Dispatch+Async.swift

@@ -68,22 +68,9 @@ extension AsyncThrowingStream where Element == UInt8, Failure == any Error {
     @available(visionOS, deprecated: 2.0, message: "Use the AsyncSequence-returning overload.")
     public static func _dataStream(reading fileDescriptor: DispatchFD, on queue: SWBQueue) -> AsyncThrowingStream<Element, any Error> {
         AsyncThrowingStream { continuation in
-            let newFD: DispatchFD
-            do {
-                newFD = try fileDescriptor._duplicate()
-            } catch {
-                continuation.finish(throwing: error)
-                return
-            }
-
-            let io = SWBDispatchIO.stream(fileDescriptor: newFD, queue: queue) { error in
-                do {
-                    try newFD._close()
-                    if error != 0 {
-                        continuation.finish(throwing: POSIXError(error, context: "dataStream(reading: \(fileDescriptor))#1"))
-                    }
-                } catch {
-                    continuation.finish(throwing: error)
+            let io = SWBDispatchIO.stream(fileDescriptor: fileDescriptor, queue: queue) { error in
+                if error != 0 {
+                    continuation.finish(throwing: POSIXError(error, context: "dataStream(reading: \(fileDescriptor))#1"))
                 }
             }
             io.setLimit(lowWater: 0)
@@ -120,51 +107,15 @@ extension AsyncThrowingStream where Element == UInt8, Failure == any Error {
 extension AsyncSequence where Element == UInt8, Failure == any Error {
     /// Returns an async stream which reads bytes from the specified file descriptor. Unlike `FileHandle.bytes`, it does not block the caller.
     public static func dataStream(reading fileDescriptor: DispatchFD, on queue: SWBQueue) -> any AsyncSequence<Element, any Error> {
-        AsyncThrowingStream<SWBDispatchData, any Error> { continuation in
-            let newFD: DispatchFD
-            do {
-                newFD = try fileDescriptor._duplicate()
-            } catch {
-                continuation.finish(throwing: error)
-                return
-            }
-
-            let io = SWBDispatchIO.stream(fileDescriptor: newFD, queue: queue) { error in
-                do {
-                    try newFD._close()
-                    if error != 0 {
-                        let context = "dataStream(reading: \(fileDescriptor) \"\(Result { try fileDescriptor._filePath() })\")#1"
-                        continuation.finish(throwing: POSIXError(error, context: context))
-                    }
-                } catch {
-                    continuation.finish(throwing: error)
-                }
-            }
-            io.setLimit(lowWater: 0)
-            io.setLimit(highWater: 4096)
-
-            continuation.onTermination = { termination in
-                if case .cancelled = termination {
-                    io.close(flags: .stop)
-                } else {
-                    io.close()
-                }
-            }
-
-            io.read(offset: 0, length: .max, queue: queue) { done, data, error in
-                guard error == 0 else {
-                    let context = "dataStream(reading: \(fileDescriptor) \"\(Result { try fileDescriptor._filePath() })\")#2"
-                    continuation.finish(throwing: POSIXError(error, context: context))
-                    return
-                }
-
-                let data = data ?? .empty
-                continuation.yield(data)
-
-                if done {
-                    continuation.finish()
+        AsyncThrowingStream<SWBDispatchData, any Error> {
+            while !Task.isCancelled {
+                let chunk = try await fileDescriptor.readChunk(upToLength: 4096)
+                if chunk.isEmpty {
+                    return nil
                 }
+                return chunk
             }
+            throw CancellationError()
         }.flattened
     }
 }

Sources/SWBUtil/SWBDispatch.swift

@@ -49,33 +49,6 @@ public struct DispatchFD {
         rawValue = fileHandle.fileDescriptor
         #endif
     }
-
-    internal func _duplicate() throws -> DispatchFD {
-        #if os(Windows)
-        return self
-        #else
-        return try DispatchFD(fileDescriptor: FileDescriptor(rawValue: rawValue).duplicate())
-        #endif
-    }
-
-    internal func _close() throws {
-        #if !os(Windows)
-        try FileDescriptor(rawValue: rawValue).close()
-        #endif
-    }
-
-    // Only exists to help debug a rare concurrency issue where the file descriptor goes invalid
-    internal func _filePath() throws -> String {
-        #if canImport(Darwin)
-        var buffer = [CChar](repeating: 0, count: Int(MAXPATHLEN))
-        if fcntl(rawValue, F_GETPATH, &buffer) == -1 {
-            throw POSIXError(errno, "fcntl", String(rawValue), "F_GETPATH")
-        }
-        return String(cString: buffer)
-        #else
-        return String()
-        #endif
-    }
 }
 
 // @unchecked: rdar://130051790 (DispatchData should be Sendable)