Async tests (#108)

* Async tests

* Add swift 5.6 to CI

Author: adam-fowler

.github/workflows/ci.yml

@@ -51,6 +51,7 @@ jobs:
           - swift:5.3
           - swift:5.4
           - swift:5.5
+          - swift:5.6
     container:
       image: ${{ matrix.tag }}
     services:

Tests/MQTTNIOTests/MQTTNIOTests+async.swift

@@ -34,20 +34,6 @@ final class AsyncMQTTNIOTests: XCTestCase {
         return logger
     }()
 
-    func XCTRunAsyncAndBlock(_ closure: @escaping () async throws -> Void) {
-        let dg = DispatchGroup()
-        dg.enter()
-        Task {
-            do {
-                try await closure()
-            } catch {
-                XCTFail("\(error)")
-            }
-            dg.leave()
-        }
-        dg.wait()
-    }
-
     func createClient(identifier: String, version: MQTTClient.Version = .v3_1_1, timeout: TimeAmount? = .seconds(10)) -> MQTTClient {
         MQTTClient(
             host: Self.hostname,
@@ -59,49 +45,45 @@ final class AsyncMQTTNIOTests: XCTestCase {
         )
     }
 
-    func testConnect() {
+    func testConnect() async throws {
         let client = self.createClient(identifier: "testConnect+async")
-        self.XCTRunAsyncAndBlock {
-            try await client.connect()
-            try await client.disconnect()
-            try await client.shutdown()
-        }
+        try await client.connect()
+        try await client.disconnect()
+        try await client.shutdown()
     }
 
-    func testPublishSubscribe() {
+    func testPublishSubscribe() async throws {
         let expectation = XCTestExpectation(description: "testPublishSubscribe")
         expectation.expectedFulfillmentCount = 1
 
         let client = self.createClient(identifier: "testPublish+async")
         let client2 = self.createClient(identifier: "testPublish+async2")
         let payloadString = "Hello"
-        self.XCTRunAsyncAndBlock {
-            try await client.connect()
-            try await client2.connect()
-            _ = try await client2.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)])
-            client2.addPublishListener(named: "test") { result in
-                switch result {
-                case .success(let publish):
-                    var buffer = publish.payload
-                    let string = buffer.readString(length: buffer.readableBytes)
-                    XCTAssertEqual(string, payloadString)
-                    expectation.fulfill()
-                case .failure(let error):
-                    XCTFail("\(error)")
-                }
+        try await client.connect()
+        try await client2.connect()
+        _ = try await client2.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)])
+        client2.addPublishListener(named: "test") { result in
+            switch result {
+            case .success(let publish):
+                var buffer = publish.payload
+                let string = buffer.readString(length: buffer.readableBytes)
+                XCTAssertEqual(string, payloadString)
+                expectation.fulfill()
+            case .failure(let error):
+                XCTFail("\(error)")
             }
-            try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
+        }
+        try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
 
-            self.wait(for: [expectation], timeout: 2)
+        self.wait(for: [expectation], timeout: 2)
 
-            try await client.disconnect()
-            try await client2.disconnect()
-            try await client.shutdown()
-            try await client2.shutdown()
-        }
+        try await client.disconnect()
+        try await client2.disconnect()
+        try await client.shutdown()
+        try await client2.shutdown()
     }
 
-    func testPing() {
+    func testPing() async throws {
         let client = MQTTClient(
             host: Self.hostname,
             port: 1883,
@@ -111,15 +93,13 @@ final class AsyncMQTTNIOTests: XCTestCase {
             configuration: .init(disablePing: true)
         )
 
-        self.XCTRunAsyncAndBlock {
-            try await client.connect()
-            try await client.ping()
-            try await client.disconnect()
-            try await client.shutdown()
-        }
+        try await client.connect()
+        try await client.ping()
+        try await client.disconnect()
+        try await client.shutdown()
     }
 
-    func testAsyncSequencePublishListener() {
+    func testAsyncSequencePublishListener() async throws {
         let expectation = XCTestExpectation(description: "testAsyncSequencePublishListener")
         expectation.expectedFulfillmentCount = 2
         let finishExpectation = XCTestExpectation(description: "testAsyncSequencePublishListener.finish")
@@ -128,43 +108,41 @@ final class AsyncMQTTNIOTests: XCTestCase {
         let client = self.createClient(identifier: "testAsyncSequencePublishListener+async", version: .v5_0)
         let client2 = self.createClient(identifier: "testAsyncSequencePublishListener+async2", version: .v5_0)
 
-        self.XCTRunAsyncAndBlock {
-            try await client.connect()
-            try await client2.connect()
-            _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)])
-            let task = Task {
-                let publishListener = client2.createPublishListener()
-                for await result in publishListener {
-                    switch result {
-                    case .success(let publish):
-                        var buffer = publish.payload
-                        let string = buffer.readString(length: buffer.readableBytes)
-                        print("Received: \(string ?? "nothing")")
-                        expectation.fulfill()
-
-                    case .failure(let error):
-                        XCTFail("\(error)")
-                    }
+        try await client.connect()
+        try await client2.connect()
+        _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)])
+        let task = Task {
+            let publishListener = client2.createPublishListener()
+            for await result in publishListener {
+                switch result {
+                case .success(let publish):
+                    var buffer = publish.payload
+                    let string = buffer.readString(length: buffer.readableBytes)
+                    print("Received: \(string ?? "nothing")")
+                    expectation.fulfill()
+
+                case .failure(let error):
+                    XCTFail("\(error)")
                 }
-                finishExpectation.fulfill()
             }
-            try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: "Hello"), qos: .atLeastOnce)
-            try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: "Goodbye"), qos: .atLeastOnce)
-            try await client.disconnect()
+            finishExpectation.fulfill()
+        }
+        try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: "Hello"), qos: .atLeastOnce)
+        try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: "Goodbye"), qos: .atLeastOnce)
+        try await client.disconnect()
 
-            self.wait(for: [expectation], timeout: 5.0)
+        self.wait(for: [expectation], timeout: 5.0)
 
-            try await client2.disconnect()
-            try await client.shutdown()
-            try await client2.shutdown()
+        try await client2.disconnect()
+        try await client.shutdown()
+        try await client2.shutdown()
 
-            self.wait(for: [finishExpectation], timeout: 5.0)
+        self.wait(for: [finishExpectation], timeout: 5.0)
 
-            _ = await task.result
-        }
+        _ = await task.result
     }
 
-    func testAsyncSequencePublishSubscriptionIdListener() {
+    func testAsyncSequencePublishSubscriptionIdListener() async throws {
         let expectation = XCTestExpectation(description: "publish listener")
         let expectation2 = XCTestExpectation(description: "publish listener2")
         expectation.expectedFulfillmentCount = 3
@@ -173,38 +151,38 @@ final class AsyncMQTTNIOTests: XCTestCase {
         let client = self.createClient(identifier: "testAsyncSequencePublishSubscriptionIdListener+async", version: .v5_0)
         let client2 = self.createClient(identifier: "testAsyncSequencePublishSubscriptionIdListener+async2", version: .v5_0)
         let payloadString = "Hello"
-        self.XCTRunAsyncAndBlock {
-            try await client.connect()
-            try await client2.connect()
-            _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)], properties: [.subscriptionIdentifier(1)])
-            _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject2", qos: .atLeastOnce)], properties: [.subscriptionIdentifier(2)])
-            let task = Task {
-                let publishListener = client2.v5.createPublishListener(subscriptionId: 1)
-                for await _ in publishListener {
-                    expectation.fulfill()
-                }
+
+        try await client.connect()
+        try await client2.connect()
+        _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject", qos: .atLeastOnce)], properties: [.subscriptionIdentifier(1)])
+        _ = try await client2.v5.subscribe(to: [.init(topicFilter: "TestSubject2", qos: .atLeastOnce)], properties: [.subscriptionIdentifier(2)])
+        let task = Task {
+            let publishListener = client2.v5.createPublishListener(subscriptionId: 1)
+            for await _ in publishListener {
                 expectation.fulfill()
             }
-            let task2 = Task {
-                let publishListener = client2.v5.createPublishListener(subscriptionId: 2)
-                for await _ in publishListener {
-                    expectation2.fulfill()
-                }
+            expectation.fulfill()
+        }
+        let task2 = Task {
+            let publishListener = client2.v5.createPublishListener(subscriptionId: 2)
+            for await _ in publishListener {
                 expectation2.fulfill()
             }
-            try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
-            try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
-            try await client.publish(to: "TestSubject2", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
-            try await client.disconnect()
-            Thread.sleep(forTimeInterval: 0.5)
-            try await client2.disconnect()
-            Thread.sleep(forTimeInterval: 0.5)
-            try client.syncShutdownGracefully()
-            try client2.syncShutdownGracefully()
-
-            _ = await task.result
-            _ = await task2.result
+            expectation2.fulfill()
         }
+        try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
+        try await client.publish(to: "TestSubject", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
+        try await client.publish(to: "TestSubject2", payload: ByteBufferAllocator().buffer(string: payloadString), qos: .atLeastOnce)
+        try await client.disconnect()
+        Thread.sleep(forTimeInterval: 0.5)
+        try await client2.disconnect()
+        Thread.sleep(forTimeInterval: 0.5)
+        try client.syncShutdownGracefully()
+        try client2.syncShutdownGracefully()
+
+        _ = await task.result
+        _ = await task2.result
+
         wait(for: [expectation, expectation2], timeout: 5.0)
     }
 }