Fix inflight publish msgs (#89)

* Added test for inflight

* Fixed inflight PUBLISH messages

Only delete them if you get a malformed packet error

* Update MQTTClient.swift

Author: adam-fowler

Sources/MQTTNIO/MQTTClient.swift

@@ -629,7 +629,9 @@ internal extension MQTTClient {
         }
         .flatMapErrorThrowing { error in
             // if publish caused server to close the connection then remove from inflight array
-            if case MQTTError.serverClosedConnection = error {
+            if case MQTTError.serverDisconnection(let ack) = error,
+               ack.reason == .malformedPacket
+            {
                 self.inflight.remove(id: packet.packetId)
             }
             throw error

Tests/MQTTNIOTests/MQTTNIOTests.swift

@@ -435,6 +435,39 @@ final class MQTTNIOTests: XCTestCase {
         try client2.disconnect().wait()
     }
 
+    func testInflight() throws {
+        let expectation = XCTestExpectation(description: "testPersistentSession")
+        expectation.expectedFulfillmentCount = 1
+
+        let client = self.createClient(identifier: "testInflight")
+        defer { XCTAssertNoThrow(try client.syncShutdownGracefully()) }
+        let client2 = self.createClient(identifier: "testPersistentSession_subscriber")
+        defer { XCTAssertNoThrow(try client2.syncShutdownGracefully()) }
+        _ = try client2.connect(cleanSession: true).wait()
+        client2.addPublishListener(named: "test") { result in
+            switch result {
+            case .success:
+                expectation.fulfill()
+
+            case .failure(let error):
+                XCTFail("\(error)")
+            }
+        }
+
+        _ = try client.connect(cleanSession: true).wait()
+        _ = try client.connect(cleanSession: false).wait()
+        _ = client.publish(to: "test/Inflight", payload: ByteBuffer(string: "Flying"), qos: .exactlyOnce)
+        try client.disconnect().wait()
+
+        _ = try client2.subscribe(to: [.init(topicFilter: "test/Inflight", qos: .atLeastOnce)]).wait()
+        _ = try client.connect(cleanSession: false).wait()
+
+        wait(for: [expectation], timeout: 5.0)
+
+        try client.disconnect().wait()
+        try client2.disconnect().wait()
+    }
+
     func testSubscribeAll() throws {
         if ProcessInfo.processInfo.environment["CI"] != nil {
             return