MQTT NIO

https://github.com/swift-server-community/mqtt-nio
Admin
MQTT NIO is a Swift NIO-based MQTT v3.1.1 and v5.0 client that supports encrypted connections,...
Tokens:7,204
Snippets:74
Trust Score:8.4
License:Apache-2.0
Update:1 month ago
Show doc for...
Tokens:
Raw
### Wait for Active Subscriptions to Finish

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-sessions.md

Waits until there are no active subscriptions on the current connection or the session using `waitUntilNoActiveSubscriptions()`. Note that in this example the subscription may not be active yet when the method is called, so it may return immediately; in practice you should wait for the subscription to be active first.

```swift
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)
            }
        }
    }

    group.addTask {
        try await MQTTConnection.withConnection(
            address: .hostname("localhost"),
            session: session
        ) { connection, _ in
            await connection.waitUntilNoActiveSubscriptions()
            // This will wait until there are no active subscriptions on the connection or the session.
        }
    }
}
```

--------------------------------

### Configure custom span attribute names for MQTT tracing

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-tracing.md

Use this snippet to customize the span attribute names used by MQTT tracing. The defaults follow Open Telemetry naming conventions; this example changes the operation and destination names to `mqtt.operation` and `mqtt.topic`. Assign the configured `MQTTTracingConfiguration` to the `tracing` member of `MQTTConnectionConfiguration`.

```swift
var tracingConfig = MQTTTracingConfiguration()
tracingConfig.attributeNames.messagingOperationName = "mqtt.operation"
tracingConfig.attributeNames.messagingDestinationName = "mqtt.topic"

var configuration = MQTTConnectionConfiguration()
configuration.tracing = tracingConfig
```

--------------------------------

### Create an MQTTSession and connect with it

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-sessions.md

Creates an MQTTSession with a client ID and passes it to MQTTConnection.withConnection. The closure receives a Bool indicating whether the server has an existing session for that client ID.

```swift
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")
    }
}
```

--------------------------------

### Connect to MQTT v5 broker

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-v5.md

Establishes a connection to an MQTT v5 broker by setting the version configuration to .v5_0(). The connection closure provides the connected MQTTConnection instance.

```swift
try await MQTTConnection.withConnection(
    address: .hostname(host),
    configuration: .init(versionConfiguration: .v5_0()),
    identifier: "My V5 Client"
) { connection in
    // You are now connected to the MQTT v5 broker
}
```

--------------------------------

### Configure custom logging for connections

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

Demonstrates passing a custom logger directly to withConnection or using the withLogger task-local wrapper.

```swift
// Your custom logger
var myLogger = Logger(label: "com.example.mqtt")
myLogger.logLevel = .trace

// Either pass the logger to the `withConnection` method
try await MQTTConnection.withConnection(
    address: .hostname("mqtt.eclipse.org"),
    identifier: "My Client",
    logger: myLogger
) { connection in
    // Connection events will be logged with `myLogger`
}

// Or wrap the connection creation code within a `withLogger` closure
try await withLogger(myLogger) { _ in
    try await MQTTConnection.withConnection(
        address: .hostname("mqtt.eclipse.org"),
        identifier: "My Client"
    ) { connection in
        // Connection events will be logged with `myLogger`
    }
}
```

--------------------------------

### Connect to MQTT broker with MQTTConnection.withConnection

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/README.md

Opens a connection to an MQTT broker using MQTTConnection.withConnection. The connection is active only inside the closure; CONNECT and DISCONNECT packets are sent automatically before and after the closure executes.

```swift
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
}
```

--------------------------------

### Run Linux tests with Docker Compose

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/CONTRIBUTING.md

Execute the project's test suite in a Linux environment using the provided docker-compose configuration.

```sh
docker-compose run test
```

--------------------------------

### Connect via Unix Domain Socket

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

Connects to a local MQTT broker via a Unix Domain Socket. The path must match the socket path configured on the broker.

```swift
try await MQTTConnection.withConnection(
    address: .unixDomainSocket(path: "/path/to/broker.socket"),
    identifier: "My UDS Client"
) { connection in
    // Connected to the broker via a Unix Domain Socket
}
```

--------------------------------

### Connect to AWS IoT via WebSocket using SotoSignerV4

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-aws.md

Use this snippet to establish a WebSocket connection to AWS IoT with Signature V4 authentication. It signs a URL with AWSSigner and then passes the request URI to MQTTConnection.withConnection. Ensure your endpoint, credentials, and region are correct; the signer expires the URL in 30 minutes.

```swift
import SotoSignerV4

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
}
```

--------------------------------

### Connect with TLS using NIOSSL

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

Connects to the mosquitto test server on port 8884 using TLS. Requires the broker's root certificate, your own certificate, and private key in PEM format. The connection is established with the NIOSSL transport.

```swift
let rootCertificate = try NIOSSLCertificate.fromPEMBytes([UInt8](mosquittoCertificateText.utf8))
let myCertificate = try NIOSSLCertificate.fromPEMBytes([UInt8](myCertificateText.utf8))
let myPrivateKey = try NIOSSLPrivateKey(bytes: [UInt8](myPrivateKeyText.utf8), format: .pem)
let tlsConfiguration: TLSConfiguration? = TLSConfiguration.forClient(
    trustRoots: .certificates(rootCertificate),
    certificateChain: myCertificate.map { .certificate($0) },
    privateKey: .privateKey(myPrivateKey)
)

try await MQTTConnection.withConnection(
    address: .hostname("test.mosquitto.org", port: 8884),
    configuration: .init(transport: .tcp(tls: .enable(.niossl(tlsConfiguration)))),
    identifier: "My SSL Client"
) { connection in
    // Connected to the broker with an encrypted connection
}
```

--------------------------------

### Publish a message with MQTTConnection.publish

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/README.md

Publishes a message to a topic using MQTTConnection.publish. The payload is provided as a ByteBuffer and the QoS is set to .atLeastOnce.

```swift
try await connection.publish(
    to: "my-topics",
    payload: ByteBuffer(string: "This is the Test payload"),
    qos: .atLeastOnce
)
```

--------------------------------

### Subscribe to a topic with MQTTConnection.subscribe

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/README.md

Subscribes to a topic with MQTTConnection.subscribe, providing a closure that receives an AsyncSequence of incoming PUBLISH messages. When the closure finishes, an UNSUBSCRIBE message is automatically sent and the subscription is cleaned up.

```swift
let subscribeInfo = MQTTSubscribeInfo(topicFilter: "my-topics", qos: .atLeastOnce)
try await connection.subscribe(to: [subscribeInfo]) { subscription in
    for try await message in subscription {
        var buffer = message.payload
        let string = buffer.readString(length: buffer.readableBytes)
        // No need to filter messages, as only messages for "my-topics" are received here
        print(string)
    }
}
```

--------------------------------

### Mosquitto Unix Domain Socket Listener Configuration

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

Enables mosquitto to listen on a Unix domain socket. Add this listener option to the mosquitto configuration file.

```text
listener 0 /path/to/broker.socket
```

--------------------------------

### Connect via WebSockets

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

Configures the transport protocol to use WebSockets by setting the transport property in MQTTConnectionConfiguration.

```swift
try await MQTTConnection.withConnection(
    address: .hostname("test.mosquitto.org", port: 8080),
    configuration: .init(transport: .webSocket(.init(...))),
    identifier: "My WebSocket Client"
) { connection in
    // Connected to the broker via WebSockets
}
```

--------------------------------

### Subscribe to Topics with MQTTSession

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-sessions.md

Opens a session subscription that remains active across connections as long as the session is not expired. The subscription receives messages while a connection using the session is active, and continues to receive messages when a new connection is opened with the same session.

```swift
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)
            }
        }
    }

    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.
            // When this connection is closed, the subscription will remain active
            // and will receive messages when a new connection is opened with the same session.
        }
    }
}
```

--------------------------------

### Create a tracing span for subscription messages with withMessageSpan

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-tracing.md

Use this snippet to create a tracing span for processing each message received from a subscription. It uses `withMessageSpan` to create a child span of the current span and adds a span link to the propagated publish tracing context. By default, the span is a child of the current span; set `createChildSpan` to `true` to make it a child of the propagated publish tracing span.

```swift
try await connection.subscribe(to: [.init(topicFilter: "myTopic", qos: .atLeastOnce)]) { subscription in
    for try await message in subscription {
        try await connection.withMessageSpan(message) { span in
            try await processMessage(message)
        }
    }
}
```

--------------------------------

### Publish with MQTT v5 properties

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-v5.md

Publishes a message to the "JSONTest" topic with the contentType property set to "application/json". Subscribers using a v5.0 client will receive the property along with the payload.

```swift
_ = try await connection.v5.publish(
    to: "JSONTest",
    payload: payload,
    qos: .atLeastOnce,
    properties: [.contentType("application/json")]
)
```

--------------------------------

### Set Custom Logger with withLogger

Source: https://github.com/swift-server-community/mqtt-nio/blob/main/Sources/MQTTNIO/MQTTNIO.docc/mqttnio-sessions.md

Sets a custom logger for the session and connections by wrapping the creation code in a `withLogger` closure, which sets a task-local logger. Incoming messages and connection events will be logged with the provided logger.

```swift
var myLogger = Logger(label: "com.example.mqtt")
myLogger.logLevel = .trace

try await withLogger(myLogger) { _ in
    let session = MQTTSession(clientID: "My Client")

    await withThrowingTaskGroup { group in
        group.addTask {
            try await session.subscribe(...) { subscription in
                for try await message in subscription {
                    // Incoming messages will be logged with `myLogger`
                }
            }
        }

        group.addTask {
            try await MQTTConnection.withConnection(
                address: .hostname("localhost"),
                session: session
            ) {
                // Connection events will be logged with `myLogger`
            }
        }
    }
}
```
Tokens:
Raw