michael/swift-mqtt-dewpoint
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: Working on async integrations.

Browse Source
This commit is contained in:
Michael Housh
2024-11-08 11:05:07 -05:00
parent 408e0484cd
commit e6d1d4578d
8 changed files with 245 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
Tests/ClientTests/AsyncClientTests.swift
View File

@@ -1,22 +1,24 @@
import XCTest
@testable import ClientLive
import EnvVars
import Logging
import Models
@testable import ClientLive
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.init(
let envVars = EnvVars(
appEnv: .testing,
host: Self.hostname,
port: "1883",
@@ -26,13 +28,13 @@ final class AsyncClientTests: XCTestCase {
)
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()
@@ -42,4 +44,73 @@ final class AsyncClientTests: XCTestCase {
try await client.publishSensor(.mixed(.init(temperature: 72.123, humidity: 50.5, needsProcessed: true)))
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 }
}

78
Tests/ClientTests/ClientTests.swift
View File

@@ -12,7 +12,7 @@ import XCTest
final class ClientLiveTests: XCTestCase {
static let hostname = ProcessInfo.processInfo.environment["MOSQUITTO_SERVER"] ?? "localhost"
let topics = Topics()
// func test_mqtt_subscription() throws {
// let mqttClient = createMQTTClient(identifier: "test_subscription")
// _ = try mqttClient.connect().wait()
@@ -23,102 +23,101 @@ final class ClientLiveTests: XCTestCase {
// try mqttClient.disconnect().wait()
// try mqttClient.syncShutdownGracefully()
// }
func test_mqtt_listener() throws {
let lock = Lock()
var publishRecieved: [MQTTPublishInfo] = []
let payloadString = "test"
let payload = ByteBufferAllocator().buffer(string: payloadString)
let client = self.createMQTTClient(identifier: "testMQTTListener_publisher")
let client = createMQTTClient(identifier: "testMQTTListener_publisher")
_ = try client.connect().wait()
client.addPublishListener(named: "test") { result in
switch result {
case .success(let publish):
case let .success(publish):
var buffer = publish.payload
let string = buffer.readString(length: buffer.readableBytes)
XCTAssertEqual(string, payloadString)
lock.withLock {
publishRecieved.append(publish)
}
case .failure(let error):
case let .failure(error):
XCTFail("\(error)")
}
}
try client.publish(to: "testMQTTSubscribe", payload: payload, qos: .atLeastOnce, retain: true).wait()
let sub = try client.v5.subscribe(to: [.init(topicFilter: "testMQTTSubscribe", qos: .atLeastOnce)]).wait()
XCTAssertEqual(sub.reasons[0], .grantedQoS1)
Thread.sleep(forTimeInterval: 2)
lock.withLock {
XCTAssertEqual(publishRecieved.count, 1)
}
try client.disconnect().wait()
try client.syncShutdownGracefully()
}
func test_client2_returnTemperature_listener() throws {
let mqttClient = createMQTTClient(identifier: "return-temperature-tests")
let state = State()
let topics = Topics()
let client = Client.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 mqttClient.disconnect().wait()
try mqttClient.syncShutdownGracefully()
// 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 = Client.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 mqttClient.disconnect().wait()
try mqttClient.syncShutdownGracefully()
// try client.shutdown().wait()
}
// func test_fetch_humidity() throws {
// let lock = Lock()
// let publishClient = createMQTTClient(identifier: "publishHumidity")
@@ -144,19 +143,20 @@ final class ClientLiveTests: XCTestCase {
// try mqttClient.disconnect().wait()
// try mqttClient.syncShutdownGracefully()
// }
// MARK: - Helpers
func createMQTTClient(identifier: String) -> MQTTNIO.MQTTClient {
MQTTNIO.MQTTClient(
host: Self.hostname,
port: 1883,
identifier: identifier,
eventLoopGroupProvider: .shared(eventLoopGroup),
logger: self.logger,
configuration: .init(version: .v5_0)
)
host: Self.hostname,
port: 1883,
identifier: identifier,
eventLoopGroupProvider: .shared(eventLoopGroup),
logger: logger,
configuration: .init(version: .v5_0)
)
}
// func createWebSocketClient(identifier: String) -> MQTTNIO.MQTTClient {
// MQTTNIO.MQTTClient(
// host: Self.hostname,
@@ -167,7 +167,7 @@ final class ClientLiveTests: XCTestCase {
// configuration: .init(useWebSockets: true, webSocketURLPath: "/mqtt")
// )
// }
// Uses default topic names.
// func createClient(mqttClient: MQTTNIO.MQTTClient, autoConnect: Bool = true) throws -> Client.MQTTClient {
// if autoConnect {
@@ -175,12 +175,12 @@ final class ClientLiveTests: XCTestCase {
// }
// return .live(client: mqttClient, topics: .init())
// }
let logger: Logger = {
var logger = Logger(label: "MQTTTests")
logger.logLevel = .trace
return logger
var logger = Logger(label: "MQTTTests")
logger.logLevel = .trace
return logger
}()
let eventLoopGroup = MultiThreadedEventLoopGroup.init(numberOfThreads: 1)
let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
}