feat: Working on async integrations.

Sources/Bootstrap/Bootstrap.swift

@@ -1,8 +1,8 @@
 import ClientLive
 import DewPointEnvironment
 import EnvVars
-import Logging
 import Foundation
+import Logging
 import Models
 import MQTTNIO
 import NIO
@@ -18,7 +18,6 @@ public func bootstrap(
   logger: Logger? = nil,
   autoConnect: Bool = true
 ) -> EventLoopFuture<DewPointEnvironment> {
-  
   logger?.debug("Bootstrapping Dew Point Controller...")
 
   return loadEnvVars(eventLoopGroup: eventLoopGroup, logger: logger)
@@ -36,7 +35,6 @@ private func loadEnvVars(
   eventLoopGroup: EventLoopGroup,
   logger: Logger?
 ) -> EventLoopFuture<EnvVars> {
-  
   logger?.debug("Loading env vars...")
 
   // TODO: Need to have the env file path passed in / dynamic.
@@ -76,13 +74,13 @@ private func loadEnvVars(
 }
 
 // MARK: TODO perhaps make loading from file an option passed in when app is launched.
+
 /// Load the topics from file in application root directory at `.topics`, if available or fall back to the defualt.
 ///
 ///  - Parameters:
 ///   - eventLoopGroup: The event loop group for the application.
 ///   - logger: An optional logger for debugging.
 private func loadTopics(eventLoopGroup: EventLoopGroup, logger: Logger?) -> EventLoopFuture<Topics> {
-  
   logger?.debug("Loading topics from file...")
 
   let topicsFilePath = URL(fileURLWithPath: #file)
@@ -94,7 +92,7 @@ private func loadTopics(eventLoopGroup: EventLoopGroup, logger: Logger?) -> Even
   let decoder = JSONDecoder()
 
   // Attempt to load the topics from file in root directory.
-  let localTopics = (try? Data.init(contentsOf: topicsFilePath))
+  let localTopics = (try? Data(contentsOf: topicsFilePath))
     .flatMap { try? decoder.decode(Topics.self, from: $0) }
 
   logger?.debug(
@@ -107,20 +105,20 @@ private func loadTopics(eventLoopGroup: EventLoopGroup, logger: Logger?) -> Even
   return eventLoopGroup.next().makeSucceededFuture(localTopics ?? .init())
 }
 
-extension EventLoopFuture where Value == (EnvVars, Topics) {
+private extension EventLoopFuture where Value == (EnvVars, Topics) {
 
   /// Creates the ``DewPointEnvironment`` for the application after the ``EnvVars`` have been loaded.
   ///
   ///  - Parameters:
   ///   - eventLoopGroup: The event loop group for the application.
   ///   - logger: An optional logger for the application.
-  fileprivate func makeDewPointEnvironment(
+  func makeDewPointEnvironment(
     eventLoopGroup: EventLoopGroup,
     logger: Logger?
   ) -> EventLoopFuture<DewPointEnvironment> {
     map { envVars, topics in
       let mqttClient = MQTTClient(envVars: envVars, eventLoopGroup: eventLoopGroup, logger: logger)
-        return DewPointEnvironment.init(
+      return DewPointEnvironment(
         envVars: envVars,
         mqttClient: mqttClient,
         topics: topics
@@ -129,13 +127,13 @@ extension EventLoopFuture where Value == (EnvVars, Topics) {
   }
 }
 
-extension EventLoopFuture where Value == DewPointEnvironment {
+private extension EventLoopFuture where Value == DewPointEnvironment {
 
   /// Connects to the MQTT broker after the ``DewPointEnvironment`` has been setup.
   ///
   /// - Parameters:
   ///   - logger: An optional logger for debugging.
-  fileprivate func connectToMQTTBroker(autoConnect: Bool, logger: Logger?) -> EventLoopFuture<DewPointEnvironment> {
+  func connectToMQTTBroker(autoConnect: Bool, logger: Logger?) -> EventLoopFuture<DewPointEnvironment> {
     guard autoConnect else { return self }
     return flatMap { environment in
       logger?.debug("Connecting to MQTT Broker...")
@@ -148,9 +146,9 @@ extension EventLoopFuture where Value == DewPointEnvironment {
   }
 }
 
-extension MQTTNIO.MQTTClient {
+private extension MQTTNIO.MQTTClient {
 
-  fileprivate convenience init(envVars: EnvVars, eventLoopGroup: EventLoopGroup, logger: Logger?) {
+  convenience init(envVars: EnvVars, eventLoopGroup: EventLoopGroup, logger: Logger?) {
     self.init(
       host: envVars.host,
       port: envVars.port != nil ? Int(envVars.port!) : nil,

Sources/ClientLive/Helpers.swift

@@ -40,6 +40,7 @@ extension RelativeHumidity: BufferInitalizable {
   }
 }
 
+// TODO: Remove below when migrated to async client.
 extension MQTTNIO.MQTTClient {
   /// Logs a failure for a given topic and error.
   func logFailure(topic: String, error: Error) {

Sources/ClientLive/SensorsClient.swift

@@ -5,7 +5,8 @@ import MQTTNIO
 import NIO
 import Psychrometrics
 
-public class AsyncClient {
+// TODO: Pass in eventLoopGroup and MQTTClient.
+public actor SensorsClient {
 
   public static let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
   public let client: MQTTClient
@@ -39,17 +40,17 @@ public class AsyncClient {
     self.sensors = sensors
   }
 
-  public func addSensor(_ sensor: TemperatureAndHumiditySensor) throws {
+  public func addSensor(_ sensor: TemperatureAndHumiditySensor) async throws {
     guard sensors.firstIndex(where: { $0.location == sensor.location }) == nil else {
       throw SensorExists()
     }
     sensors.append(sensor)
   }
 
-  public func connect() async {
+  public func connect(cleanSession: Bool = true) async {
     do {
-      try await client.connect()
+      try await client.connect(cleanSession: cleanSession)
-      client.addCloseListener(named: "AsyncClient") { [self] _ in
+      client.addCloseListener(named: "SensorsClient") { [self] _ in
         guard !self.shuttingDown else { return }
         Task {
           self.logger.debug("Connection closed.")
@@ -63,6 +64,17 @@ public class AsyncClient {
     }
   }
 
+  public func start() async throws {
+    do {
+      try await subscribeToSensors()
+      try await addSensorListeners()
+      logger.debug("Begin listening to sensors...")
+    } catch {
+      logger.trace("Error:\n\(error)")
+      throw error
+    }
+  }
+
   public func shutdown() async {
     shuttingDown = true
     try? await client.disconnect()
@@ -77,105 +89,61 @@ public class AsyncClient {
   }
 
   func addSensorListeners(qos: MQTTQoS = .exactlyOnce) async throws {
-    for sensor in sensors {
+    try await subscribeToSensors(qos: qos)
-      try await client.subscribeToSensor(sensor, qos: qos)
+    client.addPublishListener(named: "SensorsClient") { result in
-      let listener = client.createPublishListener()
+      do {
-      for await result in listener {
         switch result {
         case let .success(value):
           var buffer = value.payload
           let topic = value.topicName
-          logger.debug("Received new value for topic: \(topic)")
+          self.logger.trace("Received new value for topic: \(topic)")
 
           if topic.contains("temperature") {
             // Decode and update the temperature value
             guard let temperature = Temperature(buffer: &buffer) else {
-              logger.debug("Failed to decode temperature from buffer: \(buffer)")
+              self.logger.debug("Failed to decode temperature from buffer: \(buffer)")
               throw DecodingError()
             }
-            try sensors.update(topic: topic, keyPath: \.temperature, with: temperature)
+            try self.sensors.update(topic: topic, keyPath: \.temperature, with: temperature)
-
+            Task { try await self.publishUpdates() }
           } else if topic.contains("humidity") {
             // Decode and update the temperature value
             guard let humidity = RelativeHumidity(buffer: &buffer) else {
-              logger.debug("Failed to decode humidity from buffer: \(buffer)")
+              self.logger.debug("Failed to decode humidity from buffer: \(buffer)")
               throw DecodingError()
             }
-            try sensors.update(topic: topic, keyPath: \.humidity, with: humidity)
+            try self.sensors.update(topic: topic, keyPath: \.humidity, with: humidity)
-
+            Task { try await self.publishUpdates() }
-          } else {
-            let message = """
-            Unexpected value for topic: \(topic)
-            Expected to contain either 'temperature' or 'humidity'
-            """
-            logger.debug("\(message)")
           }
 
-          // TODO: Publish dew-point & enthalpy if needed.
-
         case let .failure(error):
-          logger.trace("Error:\n\(error)")
+          self.logger.trace("Error:\n\(error)")
           throw error
         }
+      } catch {
+        self.logger.trace("Error:\n\(error)")
       }
     }
   }
 
-  // Need to save the recieved values somewhere.
+  private func publish(double: Double?, to topic: String) async throws {
-  // TODO: Remove.
+    guard let double else { return }
-  func addPublishListener<T>(
+    let rounded = round(double * 100) / 100
-    topic: String,
+    logger.debug("Publishing \(rounded), to: \(topic)")
-    decoding _: T.Type
-  ) async throws where T: BufferInitalizable {
-    _ = try await client.subscribe(to: [.init(topicFilter: topic, qos: .atLeastOnce)])
-    Task {
-      let listener = self.client.createPublishListener()
-      for await result in listener {
-        switch result {
-        case let .success(packet):
-          var buffer = packet.payload
-          guard let value = T(buffer: &buffer) else {
-            logger.debug("Could not decode buffer: \(buffer)")
-            return
-          }
-          logger.debug("Recieved value: \(value)")
-        case let .failure(error):
-          logger.trace("Error:\n\(error)")
-        }
-      }
-    }
-  }
-
-  private func publish(string: String, to topic: String) async throws {
     try await client.publish(
       to: topic,
-      payload: ByteBufferAllocator().buffer(string: string),
+      payload: ByteBufferAllocator().buffer(string: "\(rounded)"),
-      qos: .atLeastOnce
+      qos: .exactlyOnce,
+      retain: true
     )
   }
 
-  private func publish(double: Double, to topic: String) async throws {
+  func publishUpdates() async throws {
-    let rounded = round(double * 100) / 100
+    for sensor in sensors.filter(\.needsProcessed) {
-    try await publish(string: "\(rounded)", to: topic)
+      try await publish(double: sensor.dewPoint?.rawValue, to: sensor.topics.dewPoint)
+      try await publish(double: sensor.enthalpy?.rawValue, to: sensor.topics.enthalpy)
+      try sensors.hasProcessed(sensor)
     }
-
-  func publishDewPoint(_ request: Client.SensorPublishRequest) async throws {
-    // fix
-    guard let (dewPoint, topic) = request.dewPointData(topics: .init(), units: nil) else { return }
-    try await publish(double: dewPoint.rawValue, to: topic)
-    logger.debug("Published dewpoint: \(dewPoint.rawValue), to: \(topic)")
-  }
-
-  func publishEnthalpy(_ request: Client.SensorPublishRequest) async throws {
-    // fix
-    guard let (enthalpy, topic) = request.enthalpyData(altitude: .seaLevel, topics: .init(), units: nil) else { return }
-    try await publish(double: enthalpy.rawValue, to: topic)
-    logger.debug("Publihsed enthalpy: \(enthalpy.rawValue), to: \(topic)")
-  }
-
-  public func publishSensor(_ request: Client.SensorPublishRequest) async throws {
-    try await publishDewPoint(request)
-    try await publishEnthalpy(request)
   }
 }
 
@@ -204,13 +172,22 @@ struct DecodingError: Error {}
 struct NotFoundError: Error {}
 struct SensorExists: Error {}
 
-extension TemperatureAndHumiditySensor.Topics {
+private extension TemperatureAndHumiditySensor.Topics {
   func contains(_ topic: String) -> Bool {
     temperature == topic || humidity == topic
   }
 }
 
-extension Array where Element == TemperatureAndHumiditySensor {
+// TODO: Move to dewpoint-controller/main.swift
+public extension Array where Element == TemperatureAndHumiditySensor {
+  static var live: Self {
+    TemperatureAndHumiditySensor.Location.allCases.map {
+      TemperatureAndHumiditySensor(location: $0)
+    }
+  }
+}
+
+private extension Array where Element == TemperatureAndHumiditySensor {
 
   mutating func update<V>(
     topic: String,
@@ -223,4 +200,11 @@ extension Array where Element == TemperatureAndHumiditySensor {
     self[index][keyPath: keyPath] = value
   }
 
+  mutating func hasProcessed(_ sensor: TemperatureAndHumiditySensor) throws {
+    guard let index = firstIndex(where: { $0.id == sensor.id }) else {
+      throw NotFoundError()
+    }
+    self[index].needsProcessed = false
+  }
+
 }

Sources/EnvVars/EnvVars.swift

@@ -40,7 +40,7 @@ public struct EnvVars: Codable, Equatable {
     identifier: String = "dewPoint-controller",
     userName: String? = "mqtt_user",
     password: String? = "secret!"
-  ){
+  ) {
     self.appEnv = appEnv
     self.host = host
     self.port = port

Sources/Models/Mode.swift

@@ -1,5 +1,7 @@
 import CoreUnitTypes
 
+// TODO: Remove
+
 /// Represents the different modes that the controller can be in.
 public enum Mode: Equatable {
 

Sources/Models/State.swift

@@ -1,6 +1,7 @@
 import Foundation
 import Psychrometrics
 
+// TODO: Remove
 // TODO: Make this a struct, then create a Store class that holds the state??
 public final class State {
 
@@ -50,9 +51,9 @@ public final class State {
   }
 }
 
-extension State.Sensors {
+public extension State.Sensors {
 
-  public struct TemperatureHumiditySensor<Location>: Equatable {
+  struct TemperatureHumiditySensor<Location>: Equatable {
 
     @TrackedChanges
     public var temperature: Temperature?
@@ -97,8 +98,9 @@ extension State.Sensors {
   }
 
   // MARK: - Temperature / Humidity Sensor Location Namespaces
-  public enum MixedAir { }
+
-  public enum PostCoil { }
+  enum MixedAir {}
-  public enum Return { }
+  enum PostCoil {}
-  public enum Supply { }
+  enum Return {}
+  enum Supply {}
 }

Sources/Models/TemperatureAndHumiditySensor.swift

@@ -75,6 +75,8 @@ public struct TemperatureAndHumiditySensor: Equatable, Hashable, Identifiable {
   }
 
   /// Check whether any of the sensor values have changed and need processed.
+  ///
+  /// - Note: Setting a value will set to both the temperature and humidity properties.
   public var needsProcessed: Bool {
     get { $temperature.needsProcessed || $humidity.needsProcessed }
     set {
@@ -85,9 +87,9 @@ public struct TemperatureAndHumiditySensor: Equatable, Hashable, Identifiable {
 
   /// Represents the different locations of a temperature and humidity sensor, which can
   /// be used to derive the topic to both listen and publish new values to.
-  public enum Location: String, Equatable, Hashable {
+  public enum Location: String, CaseIterable, Equatable, Hashable {
-    case mixedAir = "mixed-air"
+    case mixedAir = "mixed_air"
-    case postCoil = "post-coil"
+    case postCoil = "post_coil"
     case `return`
     case supply
   }
@@ -95,23 +97,41 @@ public struct TemperatureAndHumiditySensor: Equatable, Hashable, Identifiable {
   /// Represents the MQTT topics to listen for updated sensor values on.
   public struct Topics: Equatable, Hashable {
 
-    /// The temperature topic of the sensor.
+    /// The dew-point temperature topic for the sensor.
-    public let temperature: String
+    public let dewPoint: String
+
+    /// The enthalpy topic for the sensor.
+    public let enthalpy: String
 
     /// The humidity topic of the sensor.
     public let humidity: String
 
+    /// The temperature topic of the sensor.
+    public let temperature: String
+
     public init(
-      temperature: String,
+      dewPoint: String,
-      humidity: String
+      enthalpy: String,
+      humidity: String,
+      temperature: String
     ) {
-      self.temperature = temperature
+      self.dewPoint = dewPoint
+      self.enthalpy = enthalpy
       self.humidity = humidity
+      self.temperature = temperature
     }
 
-    init(location: TemperatureAndHumiditySensor.Location) {
+    public init(topicPrefix: String? = "frankensystem", location: TemperatureAndHumiditySensor.Location) {
-      self.temperature = "sensors/\(location.rawValue)/temperature"
+      var prefix = topicPrefix ?? ""
-      self.humidity = "sensors/\(location.rawValue)/humidity"
+      if prefix.reversed().starts(with: "/") {
+        prefix = "\(prefix.dropLast())"
+      }
+      self.init(
+        dewPoint: "\(prefix)/sensors/\(location.rawValue)_dew_point/state",
+        enthalpy: "\(prefix)/sensors/\(location.rawValue)_enthalpy/state",
+        humidity: "\(prefix)/sensors/\(location.rawValue)_humidity/state",
+        temperature: "\(prefix)/sensors/\(location.rawValue)_temperature/state"
+      )
     }
   }
 }

Sources/Models/Topics.swift

@@ -1,3 +1,5 @@
+// TODO: Remove
+
 /// A container for all the different MQTT topics that are needed by the application.
 public struct Topics: Codable, Equatable {
   /// The command topics the application can publish to.

Sources/Models/TrackedChanges.swift

@@ -52,7 +52,7 @@ public struct TrackedChanges<Value> {
     case needsProcessed
   }
 
-  /// Check whether the value needs processed.
+  /// Whether the value needs processed.
   public var needsProcessed: Bool {
     get { tracking == .needsProcessed }
     set {

Sources/TopicsLive/Live.swift

@@ -1,22 +1,23 @@
 import Models
 
 // TODO: Fix other live topics
-extension Topics {
+public extension Topics {
 
-  public static let live = Self.init(
+  static let live = Self(
     commands: .init(),
     sensors: .init(
       mixedAirSensor: .live(location: .mixedAir),
       postCoilSensor: .live(location: .postCoil),
       returnAirSensor: .live(location: .return),
-      supplyAirSensor: .live(location: .supply)),
+      supplyAirSensor: .live(location: .supply)
+    ),
     setPoints: .init(),
     states: .init()
   )
 }
 
-extension Topics.Sensors {
+private extension Topics.Sensors {
-  fileprivate enum Location: CustomStringConvertible {
+  enum Location: CustomStringConvertible {
     case mixedAir
     case postCoil
     case `return`
@@ -37,8 +38,8 @@ extension Topics.Sensors {
   }
 }
 
-extension Topics.Sensors.TemperatureAndHumiditySensor {
+private extension Topics.Sensors.TemperatureAndHumiditySensor {
-  fileprivate static func live(
+  static func live(
     prefix: String = "frankensystem",
     location: Topics.Sensors.Location
   ) -> Self {

Tests/ClientTests/AsyncClientTests.swift

@@ -1,138 +0,0 @@
-@testable import ClientLive
-import EnvVars
-import Logging
-import Models
-import MQTTNIO
-import NIO
-import Psychrometrics
-import XCTest
-
-final class AsyncClientTests: XCTestCase {
-
-  static let hostname = ProcessInfo.processInfo.environment["MOSQUITTO_SERVER"] ?? "localhost"
-
-  static let logger: Logger = {
-    var logger = Logger(label: "AsyncClientTests")
-    logger.logLevel = .trace
-    return logger
-  }()
-
-  func createClient(identifier: String) -> AsyncClient {
-    let envVars = EnvVars(
-      appEnv: .testing,
-      host: Self.hostname,
-      port: "1883",
-      identifier: identifier,
-      userName: nil,
-      password: nil
-    )
-    return .init(envVars: envVars, logger: Self.logger)
-  }
-
-  func testConnectAndShutdown() async throws {
-    let client = createClient(identifier: "testConnectAndShutdown")
-    await client.connect()
-    await client.shutdown()
-  }
-
-  func testPublishingSensor() async throws {
-    let client = createClient(identifier: "testPublishingSensor")
-    await client.connect()
-    let topic = Topics().sensors.mixedAirSensor.dewPoint
-    try await client.addPublishListener(topic: topic, decoding: Temperature.self)
-    try await client.publishSensor(.mixed(.init(temperature: 71.123, humidity: 50.5, needsProcessed: true)))
-    try await client.publishSensor(.mixed(.init(temperature: 72.123, humidity: 50.5, needsProcessed: true)))
-    await client.shutdown()
-  }
-
-  func testSensor() async throws {
-    let client = createClient(identifier: "testSensor")
-    let mqtt = client.client
-    try client.addSensor(.init(location: .mixedAir))
-    await client.connect()
-
-    Task { try await client.addSensorListeners() }
-
-    try await mqtt.publish(
-      to: "sensors/mixed-air/temperture",
-      payload: ByteBufferAllocator().buffer(string: "75.123"),
-      qos: .atLeastOnce
-    )
-
-    try await Task.sleep(for: .seconds(2))
-
-    XCTAssert(client.sensors.first!.needsProcessed)
-    XCTAssertEqual(client.sensors.first!.temperature, 75.123)
-
-    await client.shutdown()
-  }
-
-//   func testNewSensorSyntax() async throws {
-//     let client = createClient(identifier: "testNewSensorSyntax")
-//     let mqtt = client.client
-//     let receivedPublishInfo = PublishInfoContainer()
-//     let payload = ByteBufferAllocator().buffer(string: "75.123")
-//     let sensor = TemperatureAndHumiditySensor(location: .return)
-//
-//     await client.connect()
-//
-//     try await mqtt.subscribeToTemperature(sensor: sensor)
-//
-//     let listener = mqtt.createPublishListener()
-//
-//     Task { [receivedPublishInfo] in
-//       for await result in listener {
-//         switch result {
-//         case let .failure(error):
-//           XCTFail("\(error)")
-//         case let .success(publish):
-//           await receivedPublishInfo.addPublishInfo(publish)
-//         }
-//       }
-//     }
-//
-//     try await mqtt.publish(to: sensor.topics.temperature, payload: payload, qos: .atLeastOnce)
-//
-//     try await Task.sleep(for: .seconds(2))
-//
-//     XCTAssertEqual(receivedPublishInfo.count, 1)
-//
-//     if let publish = receivedPublishInfo.first {
-//       var buffer = publish.payload
-//       let string = buffer.readString(length: buffer.readableBytes)
-//       XCTAssertEqual(string, "75.123")
-//     } else {
-//       XCTFail("Did not receive any publish info.")
-//     }
-//
-//     try await mqtt.disconnect()
-//     try mqtt.syncShutdownGracefully()
-//   }
-}
-
-// MARK: Helpers for tests, some of these should be able to be removed once the AsyncClient interface is done.
-
-extension MQTTClient {
-
-  func subscribeToTemperature(sensor: TemperatureAndHumiditySensor) async throws {
-    _ = try await subscribe(to: [
-      .init(topicFilter: sensor.topics.temperature, qos: .atLeastOnce)
-    ])
-  }
-}
-
-class PublishInfoContainer {
-  private var receivedPublishInfo: [MQTTPublishInfo]
-
-  init() {
-    self.receivedPublishInfo = []
-  }
-
-  func addPublishInfo(_ info: MQTTPublishInfo) async {
-    receivedPublishInfo.append(info)
-  }
-
-  var count: Int { receivedPublishInfo.count }
-
-  var first: MQTTPublishInfo? { receivedPublishInfo.first }
-}

Tests/ClientTests/SensorsClientTests.swift

@@ -0,0 +1,116 @@
+@testable import ClientLive
+import EnvVars
+import Logging
+import Models
+import MQTTNIO
+import NIO
+import Psychrometrics
+import XCTest
+
+final class AsyncClientTests: XCTestCase {
+
+  static let hostname = ProcessInfo.processInfo.environment["MOSQUITTO_SERVER"] ?? "localhost"
+
+  static let logger: Logger = {
+    var logger = Logger(label: "AsyncClientTests")
+    logger.logLevel = .debug
+    return logger
+  }()
+
+  func createClient(identifier: String) -> SensorsClient {
+    let envVars = EnvVars(
+      appEnv: .testing,
+      host: Self.hostname,
+      port: "1883",
+      identifier: identifier,
+      userName: nil,
+      password: nil
+    )
+    return .init(envVars: envVars, logger: Self.logger)
+  }
+
+  func testConnectAndShutdown() async throws {
+    let client = createClient(identifier: "testConnectAndShutdown")
+    await client.connect()
+    await client.shutdown()
+  }
+
+  func testSensorCapturesPublishedState() async throws {
+    let client = createClient(identifier: "testSensorCapturesPublishedState")
+    let mqtt = await client.client
+    let sensor = TemperatureAndHumiditySensor(location: .mixedAir, units: .metric)
+    let publishInfo = PublishInfoContainer(topicFilters: [
+      sensor.topics.dewPoint,
+      sensor.topics.enthalpy
+    ])
+
+    try await client.addSensor(sensor)
+    await client.connect()
+    try await client.start()
+
+    _ = try await mqtt.subscribe(to: [
+      .init(topicFilter: sensor.topics.dewPoint, qos: .exactlyOnce),
+      .init(topicFilter: sensor.topics.enthalpy, qos: .exactlyOnce)
+    ])
+
+    let listener = mqtt.createPublishListener()
+    Task {
+      for await result in listener {
+        switch result {
+        case let .failure(error):
+          XCTFail("\(error)")
+        case let .success(value):
+          await publishInfo.addPublishInfo(value)
+        }
+      }
+    }
+
+    try await mqtt.publish(
+      to: sensor.topics.temperature,
+      payload: ByteBufferAllocator().buffer(string: "75.123"),
+      qos: .exactlyOnce,
+      retain: true
+    )
+
+    try await Task.sleep(for: .seconds(1))
+
+    // XCTAssert(client.sensors.first!.needsProcessed)
+    let firstSensor = await client.sensors.first!
+    XCTAssertEqual(firstSensor.temperature, .init(75.123, units: .celsius))
+
+    try await mqtt.publish(
+      to: sensor.topics.humidity,
+      payload: ByteBufferAllocator().buffer(string: "50"),
+      qos: .exactlyOnce,
+      retain: true
+    )
+
+    try await Task.sleep(for: .seconds(1))
+
+    XCTAssertEqual(publishInfo.info.count, 2)
+
+    await client.shutdown()
+  }
+}
+
+// MARK: Helpers for tests.
+
+class PublishInfoContainer {
+  private(set) var info: [MQTTPublishInfo]
+  private var topicFilters: [String]?
+
+  init(topicFilters: [String]? = nil) {
+    self.info = []
+    self.topicFilters = topicFilters
+  }
+
+  func addPublishInfo(_ info: MQTTPublishInfo) async {
+    guard let topicFilters else {
+      self.info.append(info)
+      return
+    }
+    if topicFilters.contains(info.topicName) {
+      self.info.append(info)
+    }
+  }
+}

docker-compose.yaml

@@ -13,8 +13,6 @@ services:
     working_dir: /app
     networks:
       - test
-#     volumes:
-#       - .:/app
     depends_on:
       - mosquitto-test
     environment: