Working on version 2 of the MQTT client.

Package.swift

@@ -62,7 +62,7 @@ let package = Package(
     .target(
       name: "Models",
       dependencies: [
-        .product(name: "CoreUnitTypes", package: "swift-psychrometrics"),
+        .product(name: "Psychrometrics", package: "swift-psychrometrics"),
       ]
     ),
     .target(

Sources/Client/Interface.swift

@@ -87,3 +87,41 @@ extension EventLoopFuture where Value == (Temperature, RelativeHumidity) {
     map { .init(dryBulb: $0, humidity: $1, units: units) }
   }
 }
+
+public struct Client2 {
+  
+  /// Add the publish listeners to the MQTT Broker, to be notified of published changes.
+  public var addListeners: () -> Void
+  
+  /// Connect to the MQTT Broker.
+  public var connect: () -> EventLoopFuture<Void>
+  
+  public var publishSensor: (SensorPublishRequest) -> EventLoopFuture<Void>
+  
+  /// Subscribe to appropriate topics / events.
+  public var subscribe: () -> EventLoopFuture<Void>
+  
+  /// Disconnect and close the connection to the MQTT Broker.
+  public var shutdown: () -> EventLoopFuture<Void>
+  
+  public init(
+    addListeners: @escaping () -> Void,
+    connect: @escaping () -> EventLoopFuture<Void>,
+    publishSensor: @escaping (SensorPublishRequest) -> EventLoopFuture<Void>,
+    shutdown: @escaping () -> EventLoopFuture<Void>,
+    subscribe: @escaping () -> EventLoopFuture<Void>
+  ) {
+    self.addListeners = addListeners
+    self.connect = connect
+    self.publishSensor = publishSensor
+    self.shutdown = shutdown
+    self.subscribe = subscribe
+  }
+  
+  public enum SensorPublishRequest {
+    case mixed(State.Sensors.TemperatureHumiditySensor<State.Sensors.Mixed>)
+    case postCoil(State.Sensors.TemperatureHumiditySensor<State.Sensors.PostCoil>)
+    case `return`(State.Sensors.TemperatureHumiditySensor<State.Sensors.Return>)
+    case supply(State.Sensors.TemperatureHumiditySensor<State.Sensors.Supply>)
+  }
+}

Sources/ClientLive/Live.swift

@@ -4,6 +4,7 @@ import CoreUnitTypes
 import Models
 import MQTTNIO
 import NIO
+import Psychrometrics
 
 extension Client.MQTTClient {
   
@@ -44,3 +45,245 @@ extension Client.MQTTClient {
     )
   }
 }
+
+extension Client2 {
+  
+  // The state passed in here needs to be a class or we get escaping errors in the `addListeners` method.
+  public static func live(
+    client: MQTTNIO.MQTTClient,
+    state: State,
+    topics: Topics
+  ) -> Self {
+    .init(
+      // TODO: Fix adding listeners in a more generic way.
+      addListeners: {
+//        state.addSensorListeners(to: client, topics: topics)
+        client.addPublishListener(named: topics.sensors.returnAirSensor.temperature) { result in
+          let topic = topics.sensors.returnAirSensor.temperature
+          result.logIfFailure(client: client, topic: topic)
+            .parse(as: Temperature.self)
+            .map { temperature -> () in
+              state.sensors.returnAirSensor.temperature = temperature
+            }
+        }
+        client.addPublishListener(named: topics.sensors.returnAirSensor.humidity) { result in
+          let topic = topics.sensors.returnAirSensor.humidity
+          result.logIfFailure(client: client, topic: topic)
+            .parse(as: RelativeHumidity.self)
+            .map { humidity -> () in
+              state.sensors.returnAirSensor.humidity = humidity
+            }
+        }
+      },
+      connect: {
+        client.connect()
+          .map { _ in }
+      },
+      publishSensor: { request in
+        guard let (dewPoint, topic) = request.dewPointData(topics: topics, units: state.units)
+        else {
+          client.logger.debug("No dew point for sensor.")
+          return client.eventLoopGroup.next().makeSucceededVoidFuture()
+        }
+        client.logger.debug("Publishing dew-point: \(dewPoint), to: \(topic)")
+        return client.publish(
+          to: topic,
+          payload: ByteBufferAllocator().buffer(string: "\(dewPoint.rawValue)"),
+          qos: .atLeastOnce
+        )
+        .flatMap {
+          guard let (enthalpy, topic) = request.enthalpyData(altitude: state.altitude, topics: topics, units: state.units)
+          else {
+            client.logger.debug("No enthalpy for sensor.")
+            return client.eventLoopGroup.next().makeSucceededVoidFuture()
+          }
+          client.logger.debug("Publishing enthalpy: \(enthalpy), to: \(topic)")
+          return client.publish(
+            to: topic,
+            payload: ByteBufferAllocator().buffer(string: "\(enthalpy.rawValue)"),
+            qos: .atLeastOnce
+          )
+        }
+        .map {
+          request.setHasProcessed(state: state)
+        }
+      },
+      shutdown: {
+        client.disconnect()
+          .map { try? client.syncShutdownGracefully() }
+      },
+      subscribe: {
+        // Sensor subscriptions
+        client.subscribe(to: .sensors(topics: topics))
+        .map { _ in }
+      }
+    )
+  }
+}
+
+// MARK: - Client2 Helpers.
+extension MQTTNIO.MQTTClient {
+  
+  func logFailure(topic: String, error: Error) {
+    logger.error("\(topic): \(error)")
+  }
+}
+
+extension Result where Success == MQTTPublishInfo {
+  func logIfFailure(client: MQTTNIO.MQTTClient, topic: String) -> ByteBuffer? {
+    switch self {
+    case let .success(value):
+      guard value.topicName == topic else { return nil }
+      return value.payload
+    case let .failure(error):
+      client.logFailure(topic: topic, error: error)
+      return nil
+    }
+  }
+}
+
+extension Optional where Wrapped == ByteBuffer {
+  
+  func parse<T>(as type: T.Type) -> T? where T: BufferInitalizable {
+    switch self {
+    case var .some(buffer):
+      return T.init(buffer: &buffer)
+    case .none:
+      return nil
+    }
+  }
+}
+
+struct TemperatureAndHumiditySensorKeyPathEnvelope {
+  
+  let humidityTopic: KeyPath<Topics.Sensors, String>
+  let temperatureTopic: KeyPath<Topics.Sensors, String>
+  let temperatureState: WritableKeyPath<State.Sensors, Temperature?>
+  let humidityState: WritableKeyPath<State.Sensors, RelativeHumidity?>
+  
+  func addListener(to client: MQTTNIO.MQTTClient, topics: Topics, state: State) {
+    
+    let temperatureTopic = topics.sensors[keyPath: temperatureTopic]
+    client.addPublishListener(named: temperatureTopic) { result in
+      result.logIfFailure(client: client, topic: temperatureTopic)
+        .parse(as: Temperature.self)
+        .map { temperature in
+          state.sensors[keyPath: temperatureState] = temperature
+        }
+    }
+    
+    let humidityTopic = topics.sensors[keyPath: humidityTopic]
+    client.addPublishListener(named: humidityTopic) { result in
+      result.logIfFailure(client: client, topic: humidityTopic)
+        .parse(as: RelativeHumidity.self)
+        .map { humidity in
+          state.sensors[keyPath: humidityState] = humidity
+        }
+    }
+  }
+}
+
+extension Array where Element == TemperatureAndHumiditySensorKeyPathEnvelope {
+  func addListeners(to client: MQTTNIO.MQTTClient, topics: Topics, state: State) {
+    _ = self.map { envelope in
+      envelope.addListener(to: client, topics: topics, state: state)
+    }
+  }
+}
+
+extension Array where Element == MQTTSubscribeInfo {
+  static func sensors(topics: Topics) -> Self {
+    [
+      .init(topicFilter: topics.sensors.mixedAirSensor.temperature, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.mixedAirSensor.humidity, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.postCoilSensor.temperature, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.postCoilSensor.humidity, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.returnAirSensor.temperature, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.returnAirSensor.humidity, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.supplyAirSensor.temperature, qos: .atLeastOnce),
+      .init(topicFilter: topics.sensors.supplyAirSensor.humidity, qos: .atLeastOnce),
+    ]
+  }
+}
+
+extension State {
+  func addSensorListeners(to client: MQTTNIO.MQTTClient, topics: Topics) {
+    let envelopes: [TemperatureAndHumiditySensorKeyPathEnvelope] = [
+      .init(
+        humidityTopic: \.mixedAirSensor.humidity,
+        temperatureTopic: \.mixedAirSensor.temperature,
+        temperatureState: \.mixedAirSensor.temperature,
+        humidityState: \.mixedAirSensor.humidity
+      ),
+      .init(
+        humidityTopic: \.postCoilSensor.humidity,
+        temperatureTopic: \.postCoilSensor.temperature,
+        temperatureState: \.postCoilSensor.temperature,
+        humidityState: \.postCoilSensor.humidity
+      ),
+      .init(
+        humidityTopic: \.returnAirSensor.humidity,
+        temperatureTopic: \.returnAirSensor.temperature,
+        temperatureState: \.returnAirSensor.temperature,
+        humidityState: \.returnAirSensor.humidity
+      ),
+      .init(
+        humidityTopic: \.supplyAirSensor.humidity,
+        temperatureTopic: \.supplyAirSensor.temperature,
+        temperatureState: \.supplyAirSensor.temperature,
+        humidityState: \.supplyAirSensor.humidity
+      ),
+    ]
+    envelopes.addListeners(to: client, topics: topics, state: self)
+  }
+}
+
+extension Client2.SensorPublishRequest {
+  
+  func dewPointData(topics: Topics, units: PsychrometricEnvironment.Units?) -> (DewPoint, String)? {
+    switch self {
+    case let .mixed(sensor):
+      guard let dp = sensor.dewPoint(units: units) else { return nil }
+      return (dp, topics.sensors.mixedAirSensor.dewPoint)
+    case let .postCoil(sensor):
+      guard let dp = sensor.dewPoint(units: units) else { return nil }
+      return (dp, topics.sensors.postCoilSensor.dewPoint)
+    case let .return(sensor):
+      guard let dp = sensor.dewPoint(units: units) else { return nil }
+      return (dp, topics.sensors.returnAirSensor.dewPoint)
+    case let .supply(sensor):
+      guard let dp = sensor.dewPoint(units: units) else { return nil }
+      return (dp, topics.sensors.supplyAirSensor.dewPoint)
+    }
+  }
+  
+  func enthalpyData(altitude: Length, topics: Topics, units: PsychrometricEnvironment.Units?) -> (EnthalpyOf<MoistAir>, String)? {
+    switch self {
+    case let .mixed(sensor):
+      guard let enthalpy = sensor.enthalpy(altitude: altitude, units: units) else { return nil }
+      return (enthalpy, topics.sensors.mixedAirSensor.enthalpy)
+    case let .postCoil(sensor):
+      guard let enthalpy = sensor.enthalpy(altitude: altitude, units: units) else { return nil }
+      return (enthalpy, topics.sensors.postCoilSensor.enthalpy)
+    case let .return(sensor):
+      guard let enthalpy = sensor.enthalpy(altitude: altitude, units: units) else { return nil }
+      return (enthalpy, topics.sensors.returnAirSensor.enthalpy)
+    case let .supply(sensor):
+      guard let enthalpy = sensor.enthalpy(altitude: altitude, units: units) else { return nil }
+      return (enthalpy, topics.sensors.supplyAirSensor.enthalpy)
+    }
+  }
+  
+  func setHasProcessed(state: State) {
+    switch self {
+    case .mixed:
+      state.sensors.mixedAirSensor.needsProcessed = false
+    case .postCoil:
+      state.sensors.postCoilSensor.needsProcessed = false
+    case .return:
+      state.sensors.returnAirSensor.needsProcessed = false
+    case .supply:
+      state.sensors.supplyAirSensor.needsProcessed = false
+    }
+  }
+}

Sources/Models/State.swift

@@ -1,44 +1,59 @@
 import Foundation
-import CoreUnitTypes
+import Psychrometrics
 
-public struct State: Equatable {
+// TODO: Make this a struct, then create a Store class that holds the state??
+public final class State {
   
+  public var altitude: Length
   public var sensors: Sensors
+  public var units: PsychrometricEnvironment.Units {
+    didSet {
+      PsychrometricEnvironment.shared.units = units
+    }
+  }
   
-  public init(sensors: Sensors = .init()) {
+  public init(
+    altitude: Length = .seaLevel,
+    sensors: Sensors = .init(),
+    units: PsychrometricEnvironment.Units = .imperial
+  ) {
+    self.altitude = altitude
     self.sensors = sensors
+    self.units = units
   }
   
   public struct Sensors: Equatable {
     
-    public var mixedSensor: TemperatureHumiditySensor<Mixed>
+    public var mixedAirSensor: TemperatureHumiditySensor<Mixed>
     public var postCoilSensor: TemperatureHumiditySensor<PostCoil>
-    public var returnSensor: TemperatureHumiditySensor<Return>
+    public var returnAirSensor: TemperatureHumiditySensor<Return>
-    public var supplySensor: TemperatureHumiditySensor<Supply>
+    public var supplyAirSensor: TemperatureHumiditySensor<Supply>
     
     public init(
-      mixedSensor: TemperatureHumiditySensor<Mixed> = .init(),
+      mixedAirSensor: TemperatureHumiditySensor<Mixed> = .init(),
       postCoilSensor: TemperatureHumiditySensor<PostCoil> = .init(),
-      returnSensor: TemperatureHumiditySensor<Return> = .init(),
+      returnAirSensor: TemperatureHumiditySensor<Return> = .init(),
-      supplySensor: TemperatureHumiditySensor<Supply> = .init()
+      supplyAirSensor: TemperatureHumiditySensor<Supply> = .init()
     ) {
-      self.mixedSensor = mixedSensor
+      self.mixedAirSensor = mixedAirSensor
       self.postCoilSensor = postCoilSensor
-      self.returnSensor = returnSensor
+      self.returnAirSensor = returnAirSensor
-      self.supplySensor = supplySensor
+      self.supplyAirSensor = supplyAirSensor
     }
     
     public var needsProcessed: Bool {
-      mixedSensor.needsProcessed
+      mixedAirSensor.needsProcessed
         || postCoilSensor.needsProcessed
-        || returnSensor.needsProcessed
+        || returnAirSensor.needsProcessed
-        || supplySensor.needsProcessed
+        || supplyAirSensor.needsProcessed
     }
   }
 }
 
 extension State.Sensors {
+  
   public struct TemperatureHumiditySensor<Location>: Equatable {
+    
     @TrackedChanges
     public var temperature: Temperature?
     
@@ -46,7 +61,25 @@ extension State.Sensors {
     public var humidity: RelativeHumidity?
     
     public var needsProcessed: Bool {
-      $temperature.needsProcessed || $humidity.needsProcessed
+      get { $temperature.needsProcessed || $humidity.needsProcessed }
+      set {
+        $temperature.needsProcessed = newValue
+        $humidity.needsProcessed = newValue
+      }
+    }
+    
+    public func dewPoint(units: PsychrometricEnvironment.Units? = nil) -> DewPoint? {
+      guard let temperature = temperature,
+            let humidity = humidity
+      else { return nil }
+      return .init(dryBulb: temperature, humidity: humidity, units: units)
+    }
+    
+    public func enthalpy(altitude: Length, units: PsychrometricEnvironment.Units? = nil) -> EnthalpyOf<MoistAir>? {
+      guard let temperature = temperature,
+            let humidity = humidity
+      else { return nil }
+      return .init(dryBulb: temperature, humidity: humidity, altitude: altitude, units: units)
     }
     
     public init(
@@ -59,53 +92,9 @@ extension State.Sensors {
     }
   }
   
-  // MARK: - Temperature / Humidity Sensor Locations
+  // MARK: - Temperature / Humidity Sensor Location Namespaces
   public enum Mixed { }
   public enum PostCoil { }
   public enum Return { }
   public enum Supply { }
 }
-
-// MARK: - Tracked Changes
-@propertyWrapper
-public struct TrackedChanges<Value> {
-  
-  private var tracking: TrackingState
-  private var value: Value
-  
-  public var wrappedValue: Value {
-    get { value }
-    set {
-      // fix
-      value = newValue
-    }
-  }
-  
-  public init(wrappedValue: Value, needsProcessed: Bool = false) {
-    self.value = wrappedValue
-    self.tracking = needsProcessed ? .needsProcessed : .hasProcessed
-  }
-  
-  enum TrackingState {
-    case hasProcessed
-    case needsProcessed
-  }
-  
-  public var needsProcessed: Bool {
-    get { tracking == .needsProcessed }
-    set {
-      if newValue {
-        tracking = .needsProcessed
-      } else {
-        tracking = .hasProcessed
-      }
-    }
-  }
-  
-  public var projectedValue: Self {
-    get { self }
-    set { self = newValue }
-  }
-}
-
-extension TrackedChanges: Equatable where Value: Equatable { }

Sources/Models/Topics.swift

@@ -38,20 +38,20 @@ public struct Topics: Codable, Equatable {
     
     public var mixedAirSensor: TemperatureAndHumiditySensor<State.Sensors.Mixed>
     public var postCoilSensor: TemperatureAndHumiditySensor<State.Sensors.PostCoil>
-    public var returnSensor: TemperatureAndHumiditySensor<State.Sensors.Return>
+    public var returnAirSensor: TemperatureAndHumiditySensor<State.Sensors.Return>
-    public var supplySensor: TemperatureAndHumiditySensor<State.Sensors.Supply>
+    public var supplyAirSensor: TemperatureAndHumiditySensor<State.Sensors.Supply>
     
     // TODO: Fix defaults.
     public init(
       mixedAirSensor: TemperatureAndHumiditySensor<State.Sensors.Mixed> = .init(),
       postCoilSensor: TemperatureAndHumiditySensor<State.Sensors.PostCoil> = .init(),
-      returnSensor: TemperatureAndHumiditySensor<State.Sensors.Return> = .init(),
+      returnAirSensor: TemperatureAndHumiditySensor<State.Sensors.Return> = .init(),
-      supplySensor: TemperatureAndHumiditySensor<State.Sensors.Supply> = .init()
+      supplyAirSensor: TemperatureAndHumiditySensor<State.Sensors.Supply> = .init()
     ) {
       self.mixedAirSensor = mixedAirSensor
       self.postCoilSensor = postCoilSensor
-      self.returnSensor = returnSensor
+      self.returnAirSensor = returnAirSensor
-      self.supplySensor = supplySensor
+      self.supplyAirSensor = supplyAirSensor
     }
     
 //    /// The temperature sensor topic.

Sources/Models/TrackedChanges.swift

@@ -0,0 +1,64 @@
+
+@propertyWrapper
+public struct TrackedChanges<Value> {
+  
+  private var tracking: TrackingState
+  private var value: Value
+  private var isEqual: (Value, Value) -> Bool
+  
+  public var wrappedValue: Value {
+    get { value }
+    set {
+      // Check if the new value is equal to the old value.
+      guard !isEqual(newValue, value) else { return }
+      // If it's not equal then set it, as well as set the tracking to `.needsProcessed`.
+      value = newValue
+      tracking = .needsProcessed
+    }
+  }
+  
+  public init(
+    wrappedValue: Value,
+    needsProcessed: Bool = false,
+    isEqual: @escaping (Value, Value) -> Bool
+  ) {
+    self.value = wrappedValue
+    self.tracking = needsProcessed ? .needsProcessed : .hasProcessed
+    self.isEqual = isEqual
+  }
+  
+  enum TrackingState {
+    case hasProcessed
+    case needsProcessed
+  }
+  
+  public var needsProcessed: Bool {
+    get { tracking == .needsProcessed }
+    set {
+      if newValue {
+        tracking = .needsProcessed
+      } else {
+        tracking = .hasProcessed
+      }
+    }
+  }
+  
+  public var projectedValue: Self {
+    get { self }
+    set { self = newValue }
+  }
+}
+
+extension TrackedChanges: Equatable where Value: Equatable {
+  public static func == (lhs: TrackedChanges<Value>, rhs: TrackedChanges<Value>) -> Bool {
+    lhs.wrappedValue == rhs.wrappedValue
+      && lhs.needsProcessed == rhs.needsProcessed
+  }
+  
+  public init(
+    wrappedValue: Value,
+    needsProcessed: Bool = false
+  ) {
+    self.init(wrappedValue: wrappedValue, needsProcessed: needsProcessed, isEqual: ==)
+  }
+}

Sources/dewPoint-controller/main.swift

@@ -23,8 +23,8 @@ if environment.envVars.appEnv == .production {
 }
 
 let relay = Relay(topic: environment.topics.commands.relays.dehumidification1)
-let tempSensor = Sensor<Temperature>(topic: environment.topics.sensors.returnSensor.temperature)
+let tempSensor = Sensor<Temperature>(topic: environment.topics.sensors.returnAirSensor.temperature)
-let humiditySensor = Sensor<RelativeHumidity>(topic: environment.topics.sensors.returnSensor.humidity)
+let humiditySensor = Sensor<RelativeHumidity>(topic: environment.topics.sensors.returnAirSensor.humidity)
 
 defer {
   logger.debug("Disconnecting")
@@ -51,7 +51,7 @@ while true {
 
   try environment.mqttClient.publish(
     dewPoint: dp,
-    to: environment.topics.sensors.returnSensor.dewPoint
+    to: environment.topics.sensors.returnAirSensor.dewPoint
   ).wait()
 
   logger.debug("Published dew point...")

Tests/ClientTests/ClientTests.swift

@@ -61,6 +61,60 @@ final class ClientLiveTests: XCTestCase {
     
   }
   
+  func test_client2_returnTemperature_listener() throws {
+    let mqttClient = createMQTTClient(identifier: "return-temperature-tests")
+    let state = State()
+    let topics = Topics()
+    let client = Client2.live(client: mqttClient, state: state, topics: topics)
+    
+    client.addListeners()
+    try client.connect().wait()
+    try client.subscribe().wait()
+    
+    _ = try mqttClient.publish(
+      to: topics.sensors.returnAirSensor.temperature,
+      payload: ByteBufferAllocator().buffer(string: "75.1234"),
+      qos: .atLeastOnce
+    ).wait()
+    
+    Thread.sleep(forTimeInterval: 2)
+    
+    XCTAssertEqual(state.sensors.returnAirSensor.temperature, .celsius(75.1234))
+    
+    try client.shutdown().wait()
+  }
+  
+  func test_client2_returnSensor_publish() throws {
+    let mqttClient = createMQTTClient(identifier: "return-temperature-tests")
+    let state = State()
+    let topics = Topics()
+    let client = Client2.live(client: mqttClient, state: state, topics: topics)
+    
+    client.addListeners()
+    try client.connect().wait()
+    try client.subscribe().wait()
+    
+    _ = try mqttClient.publish(
+      to: topics.sensors.returnAirSensor.temperature,
+      payload: ByteBufferAllocator().buffer(string: "75.1234"),
+      qos: .atLeastOnce
+    ).wait()
+    
+    _ = try mqttClient.publish(
+      to: topics.sensors.returnAirSensor.humidity,
+      payload: ByteBufferAllocator().buffer(string: "\(50.0)"),
+      qos: .atLeastOnce
+    ).wait()
+    
+    Thread.sleep(forTimeInterval: 2)
+    XCTAssert(state.sensors.returnAirSensor.needsProcessed)
+    
+    try client.publishSensor(.return(state.sensors.returnAirSensor)).wait()
+    XCTAssertFalse(state.sensors.returnAirSensor.needsProcessed)
+    
+    try client.shutdown().wait()
+  }
+  
 //  func test_fetch_humidity() throws {
 //    let lock = Lock()
 //    let publishClient = createMQTTClient(identifier: "publishHumidity")