MQTT NIO

MQTT sessions allow a client to maintain state (such as subscriptions and pending QoS 1/2 messages) across multiple connections to a broker. In MQTTNIO, sessions are managed via the MQTTSession type.

To use a session, create an MQTTSession with a unique clientID and pass it to the MQTTConnection.withConnection method. When connecting, the closure provided to withConnection receives a Bool parameter (sessionPresent) which indicates whether the server already has an existing session for that specific client ID.

let session = MQTTSession(clientID: "My Client")

try await MQTTConnection.withConnection(
    address: .hostname("localhost"),
    session: session
) { connection, sessionPresent in
    if sessionPresent {
        print("The server has an existing session for this client ID")
    } else {
        print("The server does not have an existing session for this client ID")
    }
}

In MQTT NIO, connections are managed using the withConnection methods on MQTTConnection. These methods follow a scoped pattern: you provide a closure containing your MQTT operations, and the connection remains active only for the duration of that closure.

Key behaviors:

• CONNECT packets are sent automatically before the closure executes.
• DISCONNECT packets are sent automatically after the closure returns or if an error is thrown.
• The connection is closed automatically when the closure returns.
• You can wait for the connection to be closed by the server using MQTTConnection/waitOnClose().

To connect with a clean session (where you cannot resume a previous session), use the standard withConnection variant. To resume a session, use the variant that accepts a session parameter, which also provides a Bool in the closure indicating if a previous session was successfully resumed.

try await MQTTConnection.withConnection(
    address: .hostname("mqtt.eclipse.org"),
    identifier: "My Client"
) { connection in
    // You are now connected to the MQTT broker
    // The connection will be active only inside this closure
}

Instead of subscribing via a specific connection, you can use MQTTSession.subscribe(to:subscribeProperties:unsubscribeProperties:process:). This keeps the subscription active on the broker as long as the session has not expired.

When a connection is established using that session, it will automatically receive messages for those subscriptions. Even if a connection is closed, the subscription remains active and will resume receiving messages as soon as a new connection with the same session is opened.

let session = MQTTSession(clientID: "My Client")

await withThrowingTaskGroup { group in
    group.addTask {
        try await session.subscribe(to: [.init(topicFilter: "my-topics", qos: .atLeastOnce)]) { subscription in
            for try await message in subscription {
                var buffer = message.payload
                let string = buffer.readString(length: buffer.readableBytes)
                print(string)
                // The subscription remains active even if the connection task below finishes
            }
        }
    }

    group.addTask {
        try await MQTTConnection.withConnection(
            address: .hostname("localhost"),
            session: session
        ) { connection, _ in
            // The subscription will receive messages as normal while this connection is active.
        }
    }
}

To connect to an AWS IoT broker using WebSocket authentication, you must sign your initial request using AWS Signature V4. This is typically done using the AWSSigner from the Soto library.

1. Define your AWS IoT endpoint host.
2. Create an AWSSigner with your credentials, the service name iotdata, and your target region.
3. Sign the URL for the /mqtt path.
4. Construct the requestURI by appending the signed query parameters to the path.
5. Use MQTTConnection.withConnection with a .webSocket transport configuration, passing the requestURI as the urlPath.

import SotoSignerV4
import MQTTNIO

let host = "MY_AWS_IOT_ENDPOINT.iot.eu-west-1.amazonaws.com"
let headers = HTTPHeaders([("host", host)])
let signer = AWSSigner(
    credentials: StaticCredential(accessKeyId: "MYACCESSKEY", secretAccessKey: "MYSECRETKEY"),
    name: "iotdata",
    region: "eu-west-1"
)
let signedURL = signer.signURL(
    url: URL(string: "https://\(host)/mqtt")!,
    method: .GET,
    headers: headers,
    body: .none,
    expires: .minutes(30)
)
let requestURI = "/mqtt?\(signedURL.query!)"

try await MQTTConnection.withConnection(
    address: .hostname(host),
    configuration: .init(transport: .webSocket(.init(urlPath: requestURI), tls: .enable(...))),
    identifier: "My AWS Client"
) { connection in
    // You are now connected to AWS IoT
}