SwiftOCADevice: use AsyncThrowingStream, instead of AsyncThrowingChannel

avoid a suspension point by using an AsyncThrowingStream to process inbound
messages in the CF and IORing backends

Author: lhoward

Sources/SwiftOCADevice/OCP.1/Backend/CF/Ocp1CFController.swift

@@ -89,7 +89,8 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
     _messages.eraseToAnyAsyncSequence()
   }
 
-  private var _messages = AsyncThrowingChannel<Ocp1MessageList, Error>()
+  private let _messages: AsyncThrowingStream<Ocp1MessageList, Error>
+  private let _messagesContinuation: AsyncThrowingStream<Ocp1MessageList, Error>.Continuation
   private let socket: _CFSocketWrapper
   let notificationSocket: _CFSocketWrapper
 
@@ -106,6 +107,11 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
     self.notificationSocket = notificationSocket
     peerAddress = socket.peerAddress!
 
+    (_messages, _messagesContinuation) = AsyncThrowingStream.makeStream(
+      of: Ocp1MessageList.self,
+      throwing: Error.self
+    )
+
     if peerAddress.family == AF_LOCAL {
       connectionPrefix = OcaLocalConnectionPrefix
     } else {
@@ -115,12 +121,12 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
     receiveMessageTask = Task { [self] in
       do {
         repeat {
-          try await _messages
-            .send(OcaDevice.receiveMessages { try await Array(socket.read(count: $0)) })
+          let messages = try await OcaDevice.receiveMessages { try await Array(socket.read(count: $0)) }
+          _messagesContinuation.yield(messages)
           if Task.isCancelled { break }
         } while true
       } catch {
-        _messages.fail(error)
+        _messagesContinuation.finish(throwing: error)
       }
     }
   }
@@ -133,6 +139,8 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
 
     receiveMessageTask?.cancel()
     receiveMessageTask = nil
+
+    _messagesContinuation.finish()
   }
 
   func onConnectionBecomingStale() async throws {

Sources/SwiftOCADevice/OCP.1/Backend/IORing/Ocp1IORingController.swift

@@ -80,7 +80,8 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
     _messages.eraseToAnyAsyncSequence()
   }
 
-  private var _messages = AsyncThrowingChannel<Ocp1MessageList, Error>()
+  private let _messages: AsyncThrowingStream<Ocp1MessageList, Error>
+  private let _messagesContinuation: AsyncThrowingStream<Ocp1MessageList, Error>.Continuation
   private let socket: Socket
   let notificationSocket: Socket
 
@@ -97,6 +98,11 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
     self.notificationSocket = notificationSocket
     self.endpoint = endpoint
 
+    (_messages, _messagesContinuation) = AsyncThrowingStream.makeStream(
+      of: Ocp1MessageList.self,
+      throwing: Error.self
+    )
+
     peerAddress = try AnySocketAddress(self.socket.peerAddress)
     if peerAddress.family == AF_LOCAL {
       connectionPrefix = OcaLocalConnectionPrefix
@@ -107,14 +113,15 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
     receiveMessageTask = Task { [self] in
       do {
         repeat {
-          try await _messages.send(OcaDevice.receiveMessages { try await socket.read(
+          let messages = try await OcaDevice.receiveMessages { try await socket.read(
             count: $0,
             awaitingAllRead: true
-          ) })
+          ) }
+          _messagesContinuation.yield(messages)
           if Task.isCancelled { break }
         } while true
       } catch {
-        _messages.fail(error)
+        _messagesContinuation.finish(throwing: error)
       }
     }
   }
@@ -127,6 +134,8 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
 
     receiveMessageTask?.cancel()
     receiveMessageTask = nil
+
+    _messagesContinuation.finish()
   }
 
   func onConnectionBecomingStale() async throws {