refactor reconnection logic, split out into new file

Author: lhoward

Sources/SwiftOCA/OCF/Errors.swift

@@ -34,6 +34,7 @@ public enum Ocp1Error: Error, Equatable {
   case invalidProtocolVersion
   case invalidProxyMethodResponse
   case invalidSyncValue
+  case missingKeepalive
   case noConnectionDelegate
   case noInitialValue
   case noMatchingTypeForClass

Sources/SwiftOCA/OCP.1/Backend/Ocp1CFSocketConnection.swift

@@ -401,14 +401,14 @@ public class Ocp1CFSocketConnection: Ocp1Connection {
     deviceAddress.family
   }
 
-  override func connectDevice() async throws {
+  override public func connectDevice() async throws {
     socket = try await _CFSocketWrapper(address: deviceAddress, type: type)
     try await super.connectDevice()
   }
 
-  override public func disconnectDevice(clearObjectCache: Bool) async throws {
+  override public func disconnectDevice() async throws {
     socket = nil
-    try await super.disconnectDevice(clearObjectCache: clearObjectCache)
+    try await super.disconnectDevice()
   }
 }
 

Sources/SwiftOCA/OCP.1/Backend/Ocp1FlyingSocksConnection.swift

@@ -123,7 +123,7 @@ public class Ocp1FlyingSocksConnection: Ocp1Connection {
     try? asyncSocket?.close()
   }
 
-  override func connectDevice() async throws {
+  override public func connectDevice() async throws {
     let socket = try Socket(domain: Int32(deviceAddress.family), type: socketType)
     try? setSocketOptions(socket)
     // also connect UDP sockets to ensure we do not receive unsolicited replies
@@ -135,13 +135,13 @@ public class Ocp1FlyingSocksConnection: Ocp1Connection {
     try await super.connectDevice()
   }
 
-  override public func disconnectDevice(clearObjectCache: Bool) async throws {
+  override public func disconnectDevice() async throws {
     await AsyncSocketPoolMonitor.shared.stop()
     if let asyncSocket {
       try asyncSocket.close()
       self.asyncSocket = nil
     }
-    try await super.disconnectDevice(clearObjectCache: clearObjectCache)
+    try await super.disconnectDevice()
   }
 
   public convenience init(

Sources/SwiftOCA/OCP.1/Backend/Ocp1IORingConnection.swift

@@ -79,7 +79,7 @@ public class Ocp1IORingConnection: Ocp1Connection {
     )
   }
 
-  override func connectDevice() async throws {
+  override public func connectDevice() async throws {
     let socket = try Socket(
       ring: IORing.shared,
       domain: deviceAddress.family,
@@ -94,9 +94,9 @@ public class Ocp1IORingConnection: Ocp1Connection {
     try await super.connectDevice()
   }
 
-  override public func disconnectDevice(clearObjectCache: Bool) async throws {
+  override public func disconnectDevice() async throws {
     socket = nil
-    try await super.disconnectDevice(clearObjectCache: clearObjectCache)
+    try await super.disconnectDevice()
   }
 
   fileprivate func withMappedError<T: Sendable>(

Sources/SwiftOCA/OCP.1/Ocp1Connection+Connect.swift

@@ -0,0 +1,307 @@
+//
+// Copyright (c) 2023-2024 PADL Software Pty Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the License);
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an 'AS IS' BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import AsyncAlgorithms
+import AsyncExtensions
+@preconcurrency
+import Foundation
+import Logging
+#if canImport(Combine)
+import Combine
+#elseif canImport(OpenCombine)
+import OpenCombine
+#endif
+
+private extension Ocp1Error {
+  var connectionState: Ocp1ConnectionState? {
+    switch self {
+    case .notConnected:
+      .notConnected
+    case .connectionTimeout:
+      .connectionTimedOut
+    default:
+      nil
+    }
+  }
+}
+
+private extension Error {
+  var ocp1ConnectionState: Ocp1ConnectionState {
+    (self as? Ocp1Error)?.connectionState ?? .connectionFailed
+  }
+}
+
+private extension Ocp1ConnectionState {
+  var error: Ocp1Error? {
+    switch self {
+    case .notConnected:
+      fallthrough
+    case .connectionFailed:
+      return .notConnected
+    case .connectionTimedOut:
+      return .connectionTimeout
+    default:
+      return nil
+    }
+  }
+}
+
+extension Ocp1Connection {
+  enum ReconnectionPolicy {
+    /// do not try to automatically reconnect on connection failure
+    case noReconnect
+    /// try to reconnect in the keepAlive monitor task
+    case reconnectInMonitor
+    /// try to reconnect before sending the next message
+    case reconnectOnSend
+  }
+
+  /// start receiveMessages/keepAlive monitor task
+  private func _startMonitor() {
+    let monitor = Monitor(self)
+    monitorTask = Task {
+      try await monitor.run()
+    }
+    self.monitor = monitor
+  }
+
+  /// stop receiveMessages/keepAlive monitor task
+  private func _stopMonitor() {
+    if let monitor {
+      monitor.stop()
+      self.monitor = nil
+    }
+    if let monitorTask {
+      monitorTask.cancel()
+      self.monitorTask = nil
+    }
+  }
+
+  private func _didConnectDevice() async throws {
+    _updateConnectionState(.connected)
+
+    _startMonitor()
+
+    if heartbeatTime > .zero {
+      // send keepalive to open UDP connection
+      try await sendKeepAlive()
+    }
+
+    #if canImport(Combine) || canImport(OpenCombine)
+    objectWillChange.send()
+    #endif
+
+    await refreshSubscriptions()
+    await refreshCachedObjectProperties()
+    await _refreshDeviceTreeWithPolicy()
+
+    logger.info("connected to \(self)")
+  }
+
+  private func _didDisconnectDevice(clearObjectCache: Bool) async {
+    if clearObjectCache {
+      await self.clearObjectCache()
+    }
+
+    #if canImport(Combine) || canImport(OpenCombine)
+    objectWillChange.send()
+    #endif
+
+    logger.info("disconnected from \(self)")
+  }
+
+  /// disconnect from the OCA device, retaining the object cache
+  private func _disconnectDeviceAfterConnectionFailure() async throws {
+    try await disconnectDevice()
+    await _didDisconnectDevice(clearObjectCache: false)
+  }
+
+  ///
+  /// Re-connection logic is as follows:
+  ///
+  /// * If the connection has a heartbeat, then automatic reconnection is only
+  ///   managed in the / heartbeat task
+  ///
+  /// * If the connection does not have a heartbeat, than automatic
+  ///   reconnection is managed when / sending a PDU
+  ///
+  private var _reconnectionPolicy: ReconnectionPolicy {
+    if !options.flags.contains(.automaticReconnect) {
+      .noReconnect
+    } else if heartbeatTime == .zero {
+      .reconnectOnSend
+    } else {
+      .reconnectInMonitor
+    }
+  }
+
+  /// functions with `_` prefix (with the exception of `_updateConnectionState()`) expect the
+  /// caller to update the connection state
+
+  private func _updateConnectionState(_ connectionState: Ocp1ConnectionState) {
+    _connectionState.send(connectionState)
+  }
+
+  /// connect to the OCA device, throwing `Ocp1Error.connectionTimeout` if it times out
+  private func _connectDeviceWithTimeout() async throws {
+    do {
+      try await withThrowingTimeout(of: options.connectionTimeout) {
+        try await self.connectDevice()
+      }
+    } catch Ocp1Error.responseTimeout {
+      throw Ocp1Error.connectionTimeout
+    } catch {
+      throw error
+    }
+  }
+
+  private func _refreshDeviceTreeWithPolicy() async {
+    if options.flags.contains(.refreshDeviceTreeOnConnection) {
+      logger.trace("refreshing device tree")
+      try? await refreshDeviceTree()
+    }
+  }
+
+  /// reconnect to the OCA device with exponential backoff, updating
+  /// connectionState
+  func reconnectDeviceWithBackoff() async throws {
+    var lastError: Error?
+    var backoff: Duration = options.reconnectPauseInterval
+
+    _updateConnectionState(.reconnecting)
+
+    for i in 0..<options.reconnectMaxTries {
+      do {
+        logger.trace("reconnection attempt \(i + 1)")
+        try await _connectDeviceWithTimeout()
+        try await _didConnectDevice()
+        break
+      } catch {
+        lastError = error
+        if options.reconnectExponentialBackoffThreshold.contains(i) {
+          backoff *= 2
+        }
+        logger.trace("reconnection failed with \(error), sleeping for \(backoff)")
+        try await Task.sleep(for: backoff)
+      }
+    }
+
+    if let lastError {
+      logger.trace("gave up attempting to reconnect: \(lastError)")
+      _updateConnectionState(lastError.ocp1ConnectionState)
+      throw lastError
+    } else if !isConnected {
+      logger.trace("gave up attempting to reconnect")
+      _updateConnectionState(.notConnected)
+      throw Ocp1Error.notConnected
+    }
+  }
+
+  private var _needsReconnectOnSend: Bool {
+    guard _reconnectionPolicy == .reconnectOnSend else { return false }
+
+    switch _connectionState.value {
+    case .notConnected:
+      fallthrough
+    case .connectionTimedOut:
+      fallthrough
+    case .connectionFailed:
+      return true
+    default:
+      return false
+    }
+  }
+
+  func willSendMessage() async throws {
+    guard _needsReconnectOnSend else { return }
+
+    logger.trace("reconnecting before sending message")
+    try await reconnectDeviceWithBackoff()
+  }
+
+  func didSendMessage(error: Ocp1Error? = nil) async throws {
+    if error == nil {
+      lastMessageSentTime = Monitor.now
+    }
+
+    if _reconnectionPolicy != .reconnectInMonitor, let error,
+       let connectionState = error.connectionState
+    {
+      logger
+        .trace(
+          "failed to send message: error \(error), new connection state \(connectionState); disconnecting"
+        )
+      if isConnected {
+        _updateConnectionState(connectionState)
+        try await _disconnectDeviceAfterConnectionFailure()
+      }
+    }
+  }
+
+  func didMissKeepAlive() async throws {
+    guard _connectionState.value != .reconnecting else { return }
+
+    logger.trace("missed keepalive, timing out connection with policy \(_reconnectionPolicy)")
+
+    _updateConnectionState(.connectionTimedOut)
+
+    if _reconnectionPolicy == .reconnectInMonitor {
+      try await _disconnectDeviceAfterConnectionFailure()
+      Task { try await reconnectDeviceWithBackoff() }
+    }
+
+    throw Ocp1Error.missingKeepalive
+  }
+
+  public var isConnected: Bool {
+    _connectionState.value == .connected
+  }
+}
+
+/// Public API
+public extension Ocp1Connection {
+  func connect() async throws {
+    logger.trace("connecting...")
+
+    _updateConnectionState(.connecting)
+
+    do {
+      try await _connectDeviceWithTimeout()
+    } catch {
+      logger.trace("connection failed: \(error)")
+      _updateConnectionState(error.ocp1ConnectionState)
+      throw error
+    }
+
+    let connectionState = _connectionState.value
+    if connectionState == .connecting {
+      try await _didConnectDevice()
+    } else if connectionState != .connected {
+      logger.trace("connection failed whilst attempting to connect: \(connectionState)")
+      throw connectionState.error ?? .notConnected
+    }
+  }
+
+  func disconnect() async throws {
+    await removeSubscriptions()
+
+    _updateConnectionState(.notConnected)
+
+    let clearObjectCache = !options.flags.contains(.retainObjectCacheAfterDisconnect)
+    try await disconnectDevice()
+    await _didDisconnectDevice(clearObjectCache: clearObjectCache)
+  }
+}