diff --git a/Package.swift b/Package.swift index 4c71d4e..3653fb4 100755 --- a/Package.swift +++ b/Package.swift @@ -13,11 +13,10 @@ let package = Package( .macOS(.v14) ], products: [ - .executable(name: "dewpoint-controller", targets: ["dewPoint-controller"]), + .executable(name: "dewpoint-controller", targets: ["dewpoint-controller"]), .library(name: "Models", targets: ["Models"]), .library(name: "MQTTConnectionManagerLive", targets: ["MQTTConnectionManagerLive"]), .library(name: "MQTTConnectionService", targets: ["MQTTConnectionService"]), - .library(name: "SensorsClientLive", targets: ["SensorsClientLive"]), .library(name: "SensorsService", targets: ["SensorsService"]), .library(name: "TopicDependencies", targets: ["TopicDependencies"]) ], @@ -31,11 +30,12 @@ let package = Package( ], targets: [ .executableTarget( - name: "dewPoint-controller", + name: "dewpoint-controller", dependencies: [ "Models", "MQTTConnectionManagerLive", - "SensorsClientLive", + "SensorsService", + "TopicDependencies", .product(name: "MQTTNIO", package: "mqtt-nio"), .product(name: "NIO", package: "swift-nio"), .product(name: "PsychrometricClientLive", package: "swift-psychrometrics") @@ -74,20 +74,12 @@ let package = Package( .product(name: "ServiceLifecycleTestKit", package: "swift-service-lifecycle") ] ), - .target( - name: "SensorsClientLive", - dependencies: [ - "SensorsService", - .product(name: "Dependencies", package: "swift-dependencies"), - .product(name: "MQTTNIO", package: "mqtt-nio") - ], - swiftSettings: swiftSettings - ), .target( name: "SensorsService", dependencies: [ "Models", "MQTTConnectionService", + "TopicDependencies", .product(name: "Dependencies", package: "swift-dependencies"), .product(name: "DependenciesMacros", package: "swift-dependencies"), .product(name: "MQTTNIO", package: "mqtt-nio"), diff --git a/Sources/MQTTConnectionService/MQTTConnectionService.swift b/Sources/MQTTConnectionService/MQTTConnectionService.swift index 4cea3e3..5070e7d 100644 --- a/Sources/MQTTConnectionService/MQTTConnectionService.swift +++ b/Sources/MQTTConnectionService/MQTTConnectionService.swift @@ -28,6 +28,9 @@ extension MQTTConnectionManager: TestDependencyKey { } public extension DependencyValues { + + /// A dependency that is responsible for managing the connection to + /// an MQTT broker. var mqttConnectionManager: MQTTConnectionManager { get { self[MQTTConnectionManager.self] } set { self[MQTTConnectionManager.self] = newValue } diff --git a/Sources/Models/TemperatureAndHumiditySensor.swift b/Sources/Models/TemperatureAndHumiditySensor.swift index 48a0f38..5e4d985 100644 --- a/Sources/Models/TemperatureAndHumiditySensor.swift +++ b/Sources/Models/TemperatureAndHumiditySensor.swift @@ -56,10 +56,11 @@ public struct TemperatureAndHumiditySensor: Identifiable, Sendable { public var dewPoint: DewPoint? { get async { guard let temperature = temperature, - let humidity = humidity + let humidity = humidity, + !temperature.value.isNaN, + !humidity.value.isNaN else { return nil } return try? await psychrometrics.dewPoint(.dryBulb(temperature, relativeHumidity: humidity)) - // return .init(dryBulb: temperature, humidity: humidity) } } @@ -67,12 +68,13 @@ public struct TemperatureAndHumiditySensor: Identifiable, Sendable { public var enthalpy: EnthalpyOf? { get async { guard let temperature = temperature, - let humidity = humidity + let humidity = humidity, + !temperature.value.isNaN, + !humidity.value.isNaN else { return nil } return try? await psychrometrics.enthalpy.moistAir( .dryBulb(temperature, relativeHumidity: humidity, altitude: altitude) ) - // return .init(dryBulb: temperature, humidity: humidity, altitude: altitude) } } diff --git a/Sources/SensorsClientLive/Live.swift b/Sources/SensorsClientLive/Live.swift deleted file mode 100644 index e1d665e..0000000 --- a/Sources/SensorsClientLive/Live.swift +++ /dev/null @@ -1,126 +0,0 @@ -import Dependencies -import Foundation -import MQTTNIO -import NIO -@_exported import SensorsService - -public extension SensorsClient { - - /// Creates the live implementation of the sensor client. - static func live( - client: MQTTClient, - publishQoS: MQTTQoS = .exactlyOnce, - subscribeQoS: MQTTQoS = .atLeastOnce - ) -> Self { - let listener = SensorClientListener( - client: client, - publishQoS: publishQoS, - subscribeQoS: subscribeQoS - ) - - return .init( - listen: { try await listener.listen($0) }, - publish: { try await listener.publish($0, $1) }, - shutdown: { listener.shutdown() } - ) - } -} - -struct ConnectionTimeoutError: Error {} - -private actor SensorClientListener { - - let client: MQTTClient - private let continuation: AsyncStream.Continuation - let name: String - let publishQoS: MQTTQoS - let stream: AsyncStream - let subscribeQoS: MQTTQoS - - init( - client: MQTTClient, - publishQoS: MQTTQoS, - subscribeQoS: MQTTQoS - ) { - let (stream, continuation) = AsyncStream.makeStream() - self.client = client - self.continuation = continuation - self.name = UUID().uuidString - self.publishQoS = publishQoS - self.stream = stream - self.subscribeQoS = subscribeQoS - } - - deinit { - client.logger.trace("Sensor listener is gone.") - self.client.removeCloseListener(named: name) - self.client.removePublishListener(named: name) - } - - func listen(_ topics: [String]) async throws -> AsyncStream { - client.logger.trace("Begin listen...") - // Ensure we are subscribed to the topics. - var sleepTimes = 0 - - while !client.isActive() { - guard sleepTimes < 10 else { - throw ConnectionTimeoutError() - } - try await Task.sleep(for: .milliseconds(100)) - sleepTimes += 1 - } - - client.logger.trace("Connection is active, begin listening for updates.") - client.logger.trace("Topics: \(topics)") - - _ = try await client.subscribe(to: topics.map { topic in - MQTTSubscribeInfo(topicFilter: topic, qos: subscribeQoS) - }) - - client.logger.trace("Done subscribing to topics.") - - client.addPublishListener(named: name) { result in - self.client.logger.trace("Received new result...") - switch result { - case let .failure(error): - self.client.logger.error("Received error while listening: \(error)") - case let .success(publishInfo): - // Only publish values back to caller if they are listening to a - // the topic. - if topics.contains(publishInfo.topicName) { - self.client.logger.trace("Recieved published info for: \(publishInfo.topicName)") - self.continuation.yield((buffer: publishInfo.payload, topic: publishInfo.topicName)) - } else { - self.client.logger.trace("Skipping topic: \(publishInfo.topicName)") - } - } - } - - client.addShutdownListener(named: name) { _ in - self.continuation.finish() - } - - return stream - } - - func publish(_ double: Double, _ topic: String) async throws { - // Ensure the client is active before publishing values. - guard client.isActive() else { return } - - // Round the double and publish. - let rounded = round(double * 100) / 100 - client.logger.trace("Begin publishing: \(rounded) to: \(topic)") - try await client.publish( - to: topic, - payload: ByteBufferAllocator().buffer(string: "\(rounded)"), - qos: publishQoS, - retain: true - ) - client.logger.trace("Begin publishing: \(rounded) to: \(topic)") - } - - nonisolated func shutdown() { - continuation.finish() - } - -} diff --git a/Sources/SensorsService/SensorsService.swift b/Sources/SensorsService/SensorsService.swift index 92a9626..3e65664 100644 --- a/Sources/SensorsService/SensorsService.swift +++ b/Sources/SensorsService/SensorsService.swift @@ -3,56 +3,11 @@ import DependenciesMacros import Foundation import Logging import Models +import MQTTNIO import NIO import PsychrometricClient import ServiceLifecycle - -/// Represents the interface required for the sensor service to operate. -/// -/// This allows the dependency to be controlled for testing purposes and -/// not rely on an active MQTT broker connection. -/// -/// For the live implementation see ``SensorsClientLive`` module. -/// -@DependencyClient -public struct SensorsClient: Sendable { - - public typealias PublishInfo = (buffer: ByteBuffer, topic: String) - - /// Start listening for changes to sensor values on the MQTT broker. - public var listen: @Sendable ([String]) async throws -> AsyncStream - - /// Publish dew-point or enthalpy values back to the MQTT broker. - public var publish: @Sendable (Double, String) async throws -> Void - - /// Shutdown the service. - public var shutdown: @Sendable () -> Void = {} - - /// Start listening for changes to sensor values on the MQTT broker. - public func listen(to topics: [String]) async throws -> AsyncStream { - try await listen(topics) - } - - /// Publish dew-point or enthalpy values back to the MQTT broker. - public func publish(_ value: Double, to topic: String) async throws { - try await publish(value, topic) - } -} - -extension SensorsClient: TestDependencyKey { - public static var testValue: SensorsClient { - Self() - } -} - -public extension DependencyValues { - var sensorsClient: SensorsClient { - get { self[SensorsClient.self] } - set { self[SensorsClient.self] = newValue } - } -} - -// MARK: - SensorsService +import TopicDependencies /// Service that is responsible for listening to changes of the temperature and humidity /// sensors, then publishing back the calculated dew-point temperature and enthalpy for @@ -61,12 +16,24 @@ public extension DependencyValues { /// public actor SensorsService: Service { - @Dependency(\.sensorsClient) var client - - private var sensors: [TemperatureAndHumiditySensor] + @Dependency(\.topicListener) var topicListener + @Dependency(\.topicPublisher) var topicPublisher + /// The logger to use for the service. private let logger: Logger? + /// The sensors that we are listening for updates to, so + /// that we can calculate the dew-point temperature and enthalpy + /// values to publish back to the MQTT broker. + var sensors: [TemperatureAndHumiditySensor] + + var topics: [String] { + sensors.reduce(into: [String]()) { array, sensor in + array.append(sensor.topics.temperature) + array.append(sensor.topics.humidity) + } + } + /// Create a new sensors service that listens to the passed in /// sensors. /// @@ -74,6 +41,7 @@ public actor SensorsService: Service { /// /// - Parameters: /// - sensors: The sensors to listen for changes to. + /// - logger: An optional logger to use. public init( sensors: [TemperatureAndHumiditySensor], logger: Logger? = nil @@ -85,33 +53,50 @@ public actor SensorsService: Service { /// Start the service with graceful shutdown, which will attempt to publish /// any pending changes to the MQTT broker, upon a shutdown signal. public func run() async throws { - precondition(sensors.count > 0) + precondition(sensors.count > 0, "Sensors should not be empty.") - let stream = try await client.listen(to: topics) + let stream = try await makeStream() await withGracefulShutdownHandler { await withDiscardingTaskGroup { group in - for await result in stream { + for await result in stream.cancelOnGracefulShutdown() { + logger?.trace("Received result for topic: \(result.topic)") group.addTask { await self.handleResult(result) } } + // group.cancelAll() } } onGracefulShutdown: { Task { self.logger?.trace("Received graceful shutdown.") try? await self.publishUpdates() - await self.client.shutdown() + await self.topicListener.shutdown() } } } - private var topics: [String] { - sensors.reduce(into: [String]()) { array, sensor in - array.append(sensor.topics.temperature) - array.append(sensor.topics.humidity) - } + private func makeStream() async throws -> AsyncStream<(buffer: ByteBuffer, topic: String)> { + try await topicListener.listen(to: topics) + // ignore errors, so that we continue to listen, but log them + // for debugging purposes. + .compactMap { result in + switch result { + case let .failure(error): + self.logger?.trace("Received error listening for sensors: \(error)") + return nil + case let .success(info): + return (info.payload, info.topicName) + } + } + // ignore duplicate values, to prevent publishing dew-point and enthalpy + // changes to frequently. + .removeDuplicates { lhs, rhs in + lhs.buffer == rhs.buffer + && lhs.topic == rhs.topic + } + .eraseToStream() } - private func handleResult(_ result: SensorsClient.PublishInfo) async { + private func handleResult(_ result: (buffer: ByteBuffer, topic: String)) async { do { let topic = result.topic assert(topics.contains(topic)) @@ -150,7 +135,12 @@ public actor SensorsService: Service { private func publish(_ double: Double?, to topic: String) async throws { guard let double else { return } - try await client.publish(double, to: topic) + try await topicPublisher.publish( + to: topic, + payload: ByteBufferAllocator().buffer(string: "\(double)"), + qos: .exactlyOnce, + retain: true + ) logger?.trace("Published update to topic: \(topic)") } @@ -158,6 +148,7 @@ public actor SensorsService: Service { for sensor in sensors.filter(\.needsProcessed) { try await publish(sensor.dewPoint?.value, to: sensor.topics.dewPoint) try await publish(sensor.enthalpy?.value, to: sensor.topics.enthalpy) + try sensors.hasProcessed(sensor) } } } diff --git a/Sources/TopicDependencies/TopicListener.swift b/Sources/TopicDependencies/TopicListener.swift index 31e5236..56bcbf3 100644 --- a/Sources/TopicDependencies/TopicListener.swift +++ b/Sources/TopicDependencies/TopicListener.swift @@ -10,8 +10,10 @@ import MQTTNIO @DependencyClient public struct TopicListener: Sendable { + public typealias Stream = AsyncStream> + /// Create an async stream that listens for changes to the given topics. - private var _listen: @Sendable (_ topics: [String]) async throws -> AsyncThrowingStream + private var _listen: @Sendable ([String], MQTTQoS) async throws -> Stream /// Shutdown the listener stream. public var shutdown: @Sendable () -> Void @@ -22,7 +24,7 @@ public struct TopicListener: Sendable { /// - listen: Generate an async stream of changes for the given topics. /// - shutdown: Shutdown the topic listener stream. public init( - listen: @Sendable @escaping ([String]) async throws -> AsyncThrowingStream, + listen: @Sendable @escaping ([String], MQTTQoS) async throws -> Stream, shutdown: @Sendable @escaping () -> Void ) { self._listen = listen @@ -33,16 +35,24 @@ public struct TopicListener: Sendable { /// /// - Parameters: /// - topics: The topics to listen for changes to. - public func listen(to topics: [String]) async throws -> AsyncThrowingStream { - try await _listen(topics) + /// - qos: The MQTTQoS for the subscription. + public func listen( + to topics: [String], + qos: MQTTQoS = .atLeastOnce + ) async throws -> Stream { + try await _listen(topics, qos) } /// Create an async stream that listens for changes to the given topics. /// /// - Parameters: /// - topics: The topics to listen for changes to. - public func listen(_ topics: String...) async throws -> AsyncThrowingStream { - try await listen(to: topics) + /// - qos: The MQTTQoS for the subscription. + public func listen( + _ topics: String..., + qos: MQTTQoS = .atLeastOnce + ) async throws -> Stream { + try await listen(to: topics, qos: qos) } /// Create the live implementation of the topic listener with the given MQTTClient. @@ -52,16 +62,14 @@ public struct TopicListener: Sendable { public static func live(client: MQTTClient) -> Self { let listener = MQTTTopicListener(client: client) return .init( - listen: { await listener.listen($0) }, + listen: { try await listener.listen($0, $1) }, shutdown: { listener.shutdown() } ) } } extension TopicListener: TestDependencyKey { - public static var testValue: TopicListener { - Self() - } + public static var testValue: TopicListener { Self() } } public extension DependencyValues { @@ -75,15 +83,15 @@ public extension DependencyValues { private actor MQTTTopicListener { private let client: MQTTClient - private let continuation: AsyncThrowingStream.Continuation + private let continuation: TopicListener.Stream.Continuation private let name: String - let stream: AsyncThrowingStream + let stream: TopicListener.Stream private var shuttingDown: Bool = false init( client: MQTTClient ) { - let (stream, continuation) = AsyncThrowingStream.makeStream() + let (stream, continuation) = TopicListener.Stream.makeStream() self.client = client self.continuation = continuation self.name = UUID().uuidString @@ -102,22 +110,53 @@ private actor MQTTTopicListener { continuation.finish() } client.removePublishListener(named: name) + client.removeShutdownListener(named: name) } - func listen(_ topics: [String]) async -> AsyncThrowingStream { - assert(client.isActive(), "The client is not connected.") + func listen( + _ topics: [String], + _ qos: MQTTQoS = .atLeastOnce + ) async throws(TopicListenerError) -> TopicListener.Stream { + var sleepTimes = 0 + + while !client.isActive() { + guard sleepTimes < 10 else { + throw .connectionTimeout + } + try? await Task.sleep(for: .milliseconds(100)) + sleepTimes += 1 + } + + client.logger.trace("Client is active, begin subscribing to topics.") + + let subscription = try? await client.subscribe(to: topics.map { + MQTTSubscribeInfo(topicFilter: $0, qos: qos) + }) + + guard subscription != nil else { + client.logger.error("Error subscribing to topics: \(topics)") + throw .failedToSubscribe + } + + client.logger.trace("Done subscribing, begin listening to topics.") + client.addPublishListener(named: name) { result in switch result { case let .failure(error): self.client.logger.error("Received error while listening: \(error)") - self.continuation.yield(with: .failure(error)) + self.continuation.yield(.failure(.init(error))) case let .success(publishInfo): if topics.contains(publishInfo.topicName) { self.client.logger.trace("Recieved new value for topic: \(publishInfo.topicName)") - self.continuation.yield(publishInfo) + self.continuation.yield(.success(publishInfo)) } } } + + client.addShutdownListener(named: name) { _ in + self.shutdown() + } + return stream } @@ -131,3 +170,16 @@ private actor MQTTTopicListener { Task { await self.setIsShuttingDown() } } } + +public enum TopicListenerError: Error { + case connectionTimeout + case failedToSubscribe +} + +public struct MQTTListenResultError: Error { + let underlyingError: any Error + + init(_ underlyingError: any Error) { + self.underlyingError = underlyingError + } +} diff --git a/Sources/dewPoint-controller/Application.swift b/Sources/dewPoint-controller/Application.swift index 6b8b8e4..8b5e19d 100644 --- a/Sources/dewPoint-controller/Application.swift +++ b/Sources/dewPoint-controller/Application.swift @@ -6,15 +6,16 @@ import MQTTConnectionManagerLive import MQTTNIO import NIO import PsychrometricClientLive -import SensorsClientLive +import SensorsService import ServiceLifecycle +import TopicDependencies @main struct Application { /// The main entry point of the application. static func main() async throws { - let eventloopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) + let eventloopGroup = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount) var logger = Logger(label: "dewpoint-controller") logger.logLevel = .trace @@ -35,7 +36,9 @@ struct Application { try await withDependencies { $0.psychrometricClient = .liveValue - $0.sensorsClient = .live(client: mqtt) + // $0.sensorsClient = .live(client: mqtt) + $0.topicListener = .live(client: mqtt) + $0.topicPublisher = .live(client: mqtt) $0.mqttConnectionManager = .live(client: mqtt, logger: logger) } operation: { let mqttConnection = MQTTConnectionService(cleanSession: false, logger: logger)