cancel device controller tasks on deinit, capture socket

Author: lhoward

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

@@ -118,13 +118,13 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
       connectionPrefix = OcaTcpConnectionPrefix
     }
 
-    receiveMessageTask = Task { [weak self] in
+    receiveMessageTask = Task { [weak self, socket] in
       do {
         repeat {
+          guard !Task.isCancelled else { break }
           let messages = try await OcaDevice
             .receiveMessages { try await Array(socket.read(count: $0)) }
           self?._messagesContinuation.yield(messages)
-          if Task.isCancelled { break }
         } while true
       } catch {
         self?._messagesContinuation.finish(throwing: error)
@@ -144,6 +144,12 @@ actor Ocp1CFStreamController: Ocp1CFControllerPrivate, CustomStringConvertible {
     _messagesContinuation.finish()
   }
 
+  deinit {
+    receiveMessageTask?.cancel()
+    keepAliveTask?.cancel()
+    _messagesContinuation.finish()
+  }
+
   var heartbeatTime = Duration.seconds(0) {
     didSet {
       heartbeatTimeDidChange(from: oldValue)

Sources/SwiftOCADevice/OCP.1/Backend/FlyingSocks/Ocp1FlyingFoxController.swift

@@ -85,6 +85,11 @@ actor Ocp1FlyingFoxController: Ocp1ControllerInternal, CustomStringConvertible {
     keepAliveTask = nil
   }
 
+  deinit {
+    keepAliveTask?.cancel()
+    outputStream.finish()
+  }
+
   nonisolated var identifier: String {
     String(describing: id)
   }

Sources/SwiftOCADevice/OCP.1/Backend/FlyingSocks/Ocp1FlyingSocksStreamController.swift

@@ -87,6 +87,10 @@ actor Ocp1FlyingSocksStreamController: Ocp1ControllerInternal, CustomStringConve
     keepAliveTask = nil
   }
 
+  deinit {
+    keepAliveTask?.cancel()
+  }
+
   nonisolated var identifier: String {
     address
   }

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

@@ -110,23 +110,23 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
       connectionPrefix = OcaTcpConnectionPrefix
     }
 
-    receiveMessageTask = Task { [weak self] in
+    receiveMessageTask = Task { [weak self, socket] in
       do {
         repeat {
+          guard !Task.isCancelled else { break }
           let messages = try await OcaDevice.receiveMessages { try await socket.read(
             count: $0,
             awaitingAllRead: true
           ) }
           self?._messagesContinuation.yield(messages)
-          if Task.isCancelled { break }
         } while true
       } catch {
         self?._messagesContinuation.finish(throwing: error)
       }
     }
   }
 
-  func close() async throws {
+  func close() {
     // don't close the socket, it will be closed when last reference is released
 
     keepAliveTask?.cancel()
@@ -138,6 +138,12 @@ actor Ocp1IORingStreamController: Ocp1IORingControllerPrivate, CustomStringConve
     _messagesContinuation.finish()
   }
 
+  deinit {
+    receiveMessageTask?.cancel()
+    keepAliveTask?.cancel()
+    _messagesContinuation.finish()
+  }
+
   var heartbeatTime = Duration.seconds(1) {
     didSet {
       heartbeatTimeDidChange(from: oldValue)

Sources/SwiftOCADevice/OCP.1/Backend/Local/LocalController.swift

@@ -53,4 +53,8 @@ actor OcaLocalController: Ocp1ControllerInternal {
   }
 
   func close() throws {}
+
+  deinit {
+    keepAliveTask?.cancel()
+  }
 }