### Minimal Client Setup Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Demonstrates the most basic configuration for setting up an AIOMQTT client. This is useful for getting started quickly with minimal dependencies.

```python
import asyncio
import aiomqtt

async def main():
    async with aiomqtt.open_connection("localhost") as client:
        print("Connected!")

asyncio.run(main())

```

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

### Full aiomqtt Client Configuration

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/configuration.md

This example demonstrates a complete aiomqtt client setup with all available configuration options, including SSL, Will messages, authentication, session management, keep-alive, flow control, and custom user properties. It also shows how to subscribe to topics, publish a status message, and process incoming messages.

```python
import ssl
import logging
import aiomqtt

# Complete configuration with all options
logger = logging.getLogger("mqtt")
logger.setLevel(logging.INFO)

ssl_context = ssl.create_default_context()

will = aiomqtt.Will(
    topic="home/client/status",
    payload=b"disconnected",
    qos=aiomqtt.QoS.AT_LEAST_ONCE,
    retain=True,
)

async with aiomqtt.Client(
    # Connection
    hostname="mqtt.example.com",
    port=8883,
    ssl_context=ssl_context,
    
    # Identity
    identifier="home-automation-client",
    
    # Authentication
    username="homeassistant",
    password=b"secure_password_123",
    
    # Session
    clean_start=False,
    session_expiry_interval=7200,
    
    # Will
    will=will,
    
    # Keep Alive & Reconnection
    keep_alive=60,
    reconnect=True,
    
    # Flow Control
    receive_max=1000,
    topic_alias_max=5,
    max_packet_size=65535,
    
    # Server Preferences
    request_problem_info=True,
    request_response_info=False,
    
    # Logging
    logger=logger,
    
    # Custom Properties
    user_properties=[
        ("client_version", "1.0.0"),
        ("app_name", "home_automation"),
    ],
) as client:
    
    # Subscribe to multiple topics
    await client.subscribe(
        aiomqtt.TopicFilter("home/+/temperature", qos=aiomqtt.QoS.AT_LEAST_ONCE),
        aiomqtt.TopicFilter("home/+/humidity", qos=aiomqtt.QoS.AT_LEAST_ONCE),
    )
    
    # Publish online status
    await client.publish(
        "home/client/status",
        b"connected",
        retain=True,
    )
    
    # Process messages
    async for message in client.messages():
        print(f"Received: {message.topic} = {message.payload}")
```

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

### Full Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Presents a comprehensive example showcasing a wide range of configuration options available for the AIOMQTT client, including network, security, and behavioral settings.

```python
import asyncio
import aiomqtt
import logging

logger = logging.getLogger("full_config_logger")

async def main():
    async with aiomqtt.open_connection(
        "mqtt.example.com",
        port=8883,
        login="user",
        password="password",
        tls=aiomqtt.TLSConfig(
            ca_certs="/etc/ssl/certs/ca-certificates.crt",
            certfile="/etc/ssl/certs/client.crt",
            keyfile="/etc/ssl/private/client.key"
        ),
        will=aiomqtt.Will(
            topic="clients/status",
            message="offline",
            qos=1,
            retain=True
        ),
        reconnect_interval=10,
        max_queued_messages=50,
        logger=logger
    ) as client:
        print("Connected with full configuration.")
        # ... further operations ...

asyncio.run(main())

```

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

### Install aiomqtt

Source: https://github.com/empicano/aiomqtt/blob/main/README.md

Install the aiomqtt library using pip. This command installs the latest version.

```bash
pip install aiomqtt
```

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

### Authentication Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Demonstrates how to configure username and password authentication for connecting to an MQTT broker. This is a common security measure.

```python
async with aiomqtt.open_connection(
    "localhost",
    "1883",
    login="myuser",
    password="mypassword"
) as client:
    print("Connected with authentication!")

```

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

### Establish MQTT Connection with aiomqtt

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/protocol-flow.md

Use this snippet to establish an MQTT connection. The client automatically connects upon entering the async context. Ensure you have the aiomqtt library installed. The connection parameters like hostname, port, and credentials should be configured as per your broker's setup.

```python
import aiomqtt

# Client initiates connection automatically on context enter
async with aiomqtt.Client(
    hostname="broker.example.com",
    port=1883,
    identifier="client-id",
    username="user",
    password=b"pass",
    clean_start=False,
    keep_alive=60,
) as client:
    # Connection established here
    # Background tasks (receive_loop, pingreq_loop, reconnect_loop) running
    connack = await client.connected()
    print(f"Assigned ID: {connack.assigned_client_id}")
    print(f"Receive max: {connack.receive_max}")
```

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

### Subscribe Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Demonstrates how to subscribe to MQTT topics and handle the SUBACK response. This includes specifying topics and their desired QoS levels.

```python
async with client.messages() as messages:
    await client.subscribe("topic/#")
    async for message in messages:
        print(f"Received message: {message.payload.decode()}")

```

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

### Minimal aiomqtt Client Configuration

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/configuration.md

Use this for the simplest setup with default connection parameters.

```python
import aiomqtt

# Simplest setup with defaults
async with aiomqtt.Client(hostname="test.mosquitto.org") as client:
    await client.publish("test/topic", b"Hello")
```

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

### Extended Authentication Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Provides an example of extended authentication mechanisms beyond simple username/password, such as client certificates or custom authentication plugins.

```python
# This example is conceptual as specific extended authentication methods
# depend heavily on the MQTT broker's capabilities and configuration.
# Typically involves passing additional parameters to open_connection or TLSConfig.
async with aiomqtt.open_connection("localhost", auth_plugin="my_auth_plugin") as client:
    print("Connected with extended authentication.")

```

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

### Connection Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Illustrates the sequence of messages exchanged during the MQTT connection process, starting with the CONNECT packet and ending with CONNACK.

```python
# The connection flow is handled implicitly by `aiomqtt.open_connection`.
# The following represents the conceptual flow:
# Client -> Broker: CONNECT (with protocol name, version, keep alive, etc.)
# Broker -> Client: CONNACK (with session present flag and reason code)

```

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

### TLS/SSL Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Shows how to configure the AIOMQTT client to use TLS/SSL for secure communication with the MQTT broker. This is essential for protecting data in transit.

```python
async with aiomqtt.open_connection(
    "localhost",
    tls=aiomqtt.TLSConfig(
        ca_certs="/path/to/ca.crt",
        certfile="/path/to/client.crt",
        keyfile="/path/to/client.key"
    )
) as client:
    print("Connected securely!")

```

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

### Receive QoS=0/1/2 Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Shows how to receive messages with different Quality of Service levels (0, 1, and 2) using the client's message iteration capabilities.

```python
async with client.messages() as messages:
    async for message in messages:
        print(f"Received message: Topic='{message.topic}', Payload='{message.payload.decode()}', QoS={message.qos}, MID={message.mid}")
        # For QoS 1 and 2, acknowledgments (puback, pubrec, pubrel, pubcomp) are handled internally.

```

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

### Configuration Reference

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Detailed reference for aiomqtt configuration parameters, including types, default values, and usage examples for various setups.

```APIDOC
## Configuration Reference

### Parameters

This section details all available configuration parameters for the aiomqtt client.

- **host** (str): MQTT broker hostname or IP address. Required.
- **port** (int): MQTT broker port. Default: 1883.
- **client_id** (str, optional): Unique client identifier. If not provided, a random ID is generated.
- **keepalive** (int): Maximum time interval in seconds between client and broker communications. Default: 60.
- **ssl** (bool): Enable or disable SSL/TLS. Default: False.
- **ssl_params** (dict, optional): Parameters for SSL/TLS context (e.g., `certfile`, `keyfile`, `ca_certs`).
- **loop** (asyncio.AbstractEventLoop, optional): The asyncio event loop to use. If not provided, the default loop is used.
- **clean_session** (bool): Start a clean session. Default: True.
- **protocol_version** (int): MQTT protocol version. Default: 4 (MQTT 3.1.1).
- **will_topic** (str, optional): Topic for Last Will and Testament message.
- **will_qos** (int): QoS level for the will message. Default: 0.
- **will_retain** (bool): Retain the will message. Default: False.
- **will_message** (str, optional): Payload of the will message.
- **auth** (tuple, optional): Authentication credentials as (username, password).
- **reconnect_retries** (int): Maximum reconnection attempts. -1 for infinite. Default: -1.
- **reconnect_delay** (int): Delay in seconds between reconnection attempts. Default: 1.
- **reconnect_max_delay** (int): Maximum delay in seconds between reconnection attempts. Default: 60.
- **ping_interval** (int): Interval in seconds for sending PINGREQ. 0 to disable. Default: 0.
- **ping_timeout** (int): Timeout in seconds for PINGRESP. Default: 15.
- **socket_keepalive** (bool): Enable TCP keepalive. Default: False.
- **socket_connect_timeout** (int): Timeout in seconds for socket connection. Default: 10.
- **socket_timeout** (int, optional): Timeout in seconds for socket operations. Default: None.
- **max_queued_messages** (int): Maximum number of messages in the send queue. Default: 100.
- **max_retained_messages** (int): Maximum number of retained messages to store locally. Default: 0 (unlimited).
- **extra_properties** (dict, optional): MQTT v5.0 extra properties.
- **session_expiry_interval** (int, optional): Session expiry interval in seconds (MQTT v5.0).
- **connect_properties** (dict, optional): MQTT v5.0 CONNECT properties.

### Examples

#### Minimal Setup
```python
client = aiomqtt.Client("localhost")
```

#### TLS/SSL Configuration
```python
client = aiomqtt.Client(
    "broker.example.com",
    port=8883,
    ssl=True,
    ssl_params={"ca_certs": "/path/to/ca.crt"}
)
```

#### Authentication
```python
client = aiomqtt.Client(
    "broker.example.com",
    auth=("username", "password")
)
```

#### Will Message
```python
client = aiomqtt.Client(
    "broker.example.com",
    will_topic="home/offline",
    will_message="Device is offline"
)
```

#### Auto-reconnection
```python
client = aiomqtt.Client(
    "broker.example.com",
    reconnect_retries=5,
    reconnect_delay=5
)
```
```

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

### Session Persistence Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Illustrates how to enable session persistence, allowing the client to resume a previous session upon reconnection, preserving subscriptions and QoS states.

```python
async with aiomqtt.open_connection(
    "localhost",
    will=aiomqtt.Will(
        topic="status/offline",
        message="Client disconnected",
        qos=1,
        retain=True
    )
) as client:
    print("Connected with will message configured!")

```

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

### Auto-reconnection Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Demonstrates how to enable automatic reconnection with exponential backoff. This ensures the client attempts to reconnect to the broker after a disconnection.

```python
async with aiomqtt.open_connection(
    "localhost",
    reconnect_interval=5,  # seconds
    reconnect_max_interval=60 # seconds
) as client:
    print("Client configured for auto-reconnection.")

```

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

### Flow Control Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Illustrates how to configure flow control parameters to manage the rate of message publishing and receiving, preventing buffer overflows and ensuring stability.

```python
async with aiomqtt.open_connection(
    "localhost",
    max_queued_messages=100,
    max_inflight_messages=10
) as client:
    print("Client configured with flow control.")

```

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

### QoS=1 Publish Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Shows the message exchange for publishing messages with Quality of Service level 1. This flow guarantees at least one delivery.

```python
async with client.publish("topic/qos1", payload="message", qos=1) as m:
    print(f"Published QoS 1 message: {m.mid}")

```

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

### Unix Sockets Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Shows how to configure the client to connect to an MQTT broker using Unix domain sockets instead of TCP/IP. This is often used for local inter-process communication.

```python
async with aiomqtt.open_connection(
    "unix:/path/to/mqtt.sock"
) as client:
    print("Connected via Unix socket!")

```

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

### QoS=2 Publish Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Details the three-way handshake for publishing messages with Quality of Service level 2. This flow guarantees exactly one delivery.

```python
async with client.publish("topic/qos2", payload="message", qos=2) as m:
    print(f"Published QoS 2 message: {m.mid}")

```

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

### QoS=0 Publish Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Illustrates the message exchange flow for publishing messages with Quality of Service level 0. This flow guarantees at most one delivery.

```python
async with client.publish("topic/qos0", payload="message", qos=0) as m:
    print(f"Published QoS 0 message: {m.mid}")

```

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

### Custom Logger Configuration Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Demonstrates how to provide a custom logger instance to AIOMQTT for integrating with existing logging frameworks or customizing log output.

```python
import logging

logger = logging.getLogger("my_mqtt_logger")
logger.setLevel(logging.INFO)

async with aiomqtt.open_connection(
    "localhost",
    logger=logger
) as client:
    print("Connected with custom logger.")

```

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

### Start Message Iteration Before Subscribe to Prevent Message Loss

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/INDEX.md

To avoid losing QoS=0 messages, start iterating through messages before subscribing to topics. This ensures a consumer is ready to receive messages immediately upon subscription.

```python
async for message in client.messages():  # Start this first
    pass
# Then subscribe
await client.subscribe(...)
```

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

### Subscribe to Multiple Topics with QoS

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/types.md

Example of subscribing to multiple topics using TopicFilter and different QoS levels. Ensure the client is connected before subscribing.

```python
# Subscribe to multiple topics with different QoS
await client.subscribe(
    aiomqtt.TopicFilter("ducks/louie/status", qos=aiomqtt.QoS.AT_MOST_ONCE),
    aiomqtt.TopicFilter("ducks/+/position", qos=aiomqtt.QoS.AT_LEAST_ONCE),
    aiomqtt.TopicFilter("sensors/#", qos=aiomqtt.QoS.EXACTLY_ONCE),
)
```

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

### Disconnection Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Explains the process of a client disconnecting from the broker, including the handling of the will message if configured and the connection closure.

```python
# The disconnection flow is initiated either by the client calling `client.disconnect()`
# or implicitly when the `async with aiomqtt.open_connection(...)` block exits.
# If a will message is configured, it is sent by the broker upon detecting the client's disconnection.

```

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

### Consume messages with legacy client.messages()

Source: https://github.com/empicano/aiomqtt/blob/main/docs/migration-v2.md

Example of printing all messages using the deprecated client.messages() context manager.

```python
import asyncio
import aiomqtt


async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await client.subscribe("temperature/#")
        async with client.messages() as messages:
            async for message in messages:
                print(message.payload)


asyncio.run(main())
```

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

### Implement Basic Message Routing with Regex

Source: https://github.com/empicano/aiomqtt/blob/main/docs/guides.md

Use regular expressions to define and match topic patterns for message routing. This example demonstrates compiling a pattern and testing it against various topics.

```python
import re


pattern = "ducks/+/status/#"
pattern = re.compile(pattern.replace('+', '[^/]*').replace('/#', '(|/.*)'))
cases = [
    ("ducks/louie/status", True),
    ("ducks/louie/status/", True),
    ("ducks/louie/status/foo", True),
    ("ducks/louie/location", False),
    ("dogs/louie/status", False),
    ("ducks/louie/location/status", False),
    ("ducks//louie/status", False),
    ("ducks/louie////status", False),
    ("ducks//status", True),
    ("ducks/status", False),
    ("/ducks/louie/status", False),
]
for topic, result in cases:
    assert bool(pattern.fullmatch(topic)) == result
```

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

### Subscribe to MQTT topics with aiomqtt

Source: https://github.com/empicano/aiomqtt/blob/main/README.md

Subscribe to topics and process incoming messages asynchronously. This example uses QoS level AT_MOST_ONCE.

```python
async with aiomqtt.Client(hostname="test.mosquitto.org") as client:
    await client.subscribe(
        aiomqtt.TopicFilter("ducks/#"), max_qos=aiomqtt.QoS.AT_MOST_ONCE,
    )
    async for message in client.messages():
        print(message.payload)
```

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

### Unsubscribe Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Illustrates the process of unsubscribing from MQTT topics and handling the UNSUBACK response. This is used to stop receiving messages from specific topics.

```python
await client.unsubscribe("topic/#")

```

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

### Error Handling with ConnectError

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Provides an example of how to handle connection errors that may occur during the client's attempt to establish a connection to the MQTT broker.

```python
try:
    async with aiomqtt.open_connection("nonexistent.host") as client:
        pass
except aiomqtt.exceptions.ConnectError as e:
    print(f"Connection failed: {e}")

```

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

### Handle QoS=0 Messages Before Subscribing

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/README.md

To prevent potential loss of QoS=0 messages, ensure that message iteration is started before subscribing to topics. This guarantees that any messages published while the subscription is being set up are captured.

```python
async for message in client.messages():
    # Listen for messages first
    pass
# Then subscribe
await client.subscribe(aiomqtt.TopicFilter("topic"))
```

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

### Keep-Alive Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Shows how the keep-alive mechanism works using PINGREQ and PINGRESP messages to maintain an active connection. This is crucial for long-lived connections.

```python
# The keep-alive mechanism is handled automatically by the client.
# No explicit code is typically needed here for basic usage.
# The client sends PINGREQ periodically and expects PINGRESP.

```

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

### Context Manager for Resource Management

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/implementation-details.md

Use the client as an async context manager for automatic connection and disconnection. The `__aenter__` method connects and starts background tasks, while `__aexit__` stops tasks and disconnects.

```python
async with aiomqtt.Client(...) as client:
    # __aenter__ connects and starts background tasks
    # __aexit__ stops tasks and disconnects
```

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

### __aenter__ and __aexit__

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/api-reference-client.md

The `__aenter__` method establishes a connection to the MQTT broker and starts background tasks. The `__aexit__` method cleanly disconnects from the broker, cancels background tasks, and releases resources. This allows the client to be used within an `async with` statement for simplified connection management.

```APIDOC
## `__aenter__` and `__aexit__`

### Description

Asynchronous context manager methods for managing the MQTT client's connection lifecycle. `__aenter__` connects to the broker and starts necessary background tasks, while `__aexit__` ensures a clean disconnection and resource release.

### `__aenter__`

Establishes connection to the MQTT broker and starts background tasks (receive loop, keep-alive loop, reconnect loop if enabled).

**Returns:** The Client instance itself.

**Raises:**

*   `RuntimeError`: If the client context manager is reentered (not reusable as reentrant).
*   `ProtocolError`: If the broker sends an unexpected packet type or violates the protocol.
*   `NegativeAckError`: If the CONNACK contains a non-success reason code.
*   `OSError`: If the network connection fails.

### `__aexit__`

Cleanly disconnects from the broker, cancels background tasks, and releases the reusability lock. If an exception occurred and a will message is configured, sends DISCONNECT with will message.

### Usage Example

```python
async with aiomqtt.Client(hostname="broker.example.com") as client:
    await client.publish("topic/test", b"Hello MQTT")
    await client.subscribe("topic/test")
    async for message in client.messages():
        print(message.payload)
```
```

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

### Handle aiomqtt ConnectError with Retries

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/errors.md

This example demonstrates how to implement a retry mechanism for connecting to an MQTT broker using aiomqtt. It catches ConnectError, prints an error message, and waits before retrying, with a maximum of three attempts.

```python
import asyncio
import aiomqtt

async def connect_with_retry():
    for attempt in range(3):
        try:
            async with aiomqtt.Client(hostname="broker.example.com") as client:
                # Use client
                return
        except aiomqtt.ConnectError as e:
            print(f"Connection attempt {attempt + 1} failed: {e}")
            if attempt < 2:
                await asyncio.sleep(2 ** attempt)
            else:
                raise
```

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

### Run local MQTT broker and tests

Source: https://github.com/empicano/aiomqtt/blob/main/CONTRIBUTING.md

Spin up a local Mosquitto broker using Docker and execute tests against it.

```bash
./scripts/mosquitto
```

```bash
AIOMQTT_TEST_HOSTNAME=localhost ./scripts/test
```

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

### Create and Use Will Message

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/types.md

Example of creating a Will message and passing it to the aiomqtt.Client constructor. The will message is published if the client disconnects unexpectedly.

```python
will = aiomqtt.Will(
    topic="ducks/louie/status",
    payload=b"offline",
    qos=aiomqtt.QoS.AT_LEAST_ONCE,
    retain=True,
)
async with aiomqtt.Client(hostname="broker.example.com", will=will) as client:
    await client.publish("ducks/louie/status", b"online")
```

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

### Error Handling Reference

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Reference for exceptions raised by aiomqtt, including trigger conditions, expected scenarios, and code examples for handling them.

```APIDOC
## Error Handling

### Exceptions

- **ConnectError**:
  - **Description**: Raised when the client fails to connect to the MQTT broker.
  - **Trigger Conditions**: Network issues, broker unavailable, invalid credentials, protocol errors during connection.
  - **When to Expect**: During the initial connection attempt or reconnections.
  - **Handling**: Use `try...except ConnectError` block. Implement retry logic or notify the user.

- **ProtocolError**:
  - **Description**: Raised when a protocol violation occurs.
  - **Trigger Conditions**: Malformed packets received from the broker, unexpected packet types.
  - **When to Expect**: During communication with the broker.
  - **Handling**: Log the error and potentially disconnect. This often indicates a broker issue or a bug in the library.

- **NegativeAckError**:
  - **Description**: Raised when a negative acknowledgment is received for a QoS 1 or 2 message, or for subscription/unsubscription.
  - **Attributes**: `mid`, `reason_code`, `properties`.
  - **Trigger Conditions**: Broker rejects a PUBLISH, SUBSCRIBE, or UNSUBSCRIBE operation.
  - **When to Expect**: After sending a QoS 1/2 PUBLISH, SUBSCRIBE, or UNSUBSCRIBE.
  - **Handling**: Check the `reason_code` to understand the failure and decide on a course of action (e.g., retry, log, abort).

### Exception Hierarchy

`ConnectError` and `ProtocolError` inherit from a common base exception, while `NegativeAckError` might be a separate branch or also inherit from a common base.

### Error Handling Patterns

#### Graceful Failure
Use `try...except` blocks around critical operations to catch and handle specific exceptions, allowing the application to continue running or shut down cleanly.

#### Auto-reconnection
Configure `reconnect_retries`, `reconnect_delay`, and `reconnect_max_delay` to automatically attempt reconnection after a disconnection.

#### Validated Publishing
Implement logic to handle potential `NegativeAckError` after publishing messages, especially for QoS 1 and 2, to ensure message delivery or to take corrective action.

### Code Example for Handling Connection Errors
```python
import asyncio
import aiomqtt

async def main():
    try:
        async with aiomqtt.Client("invalid.host.name") as client:
            print("Connected!")
            # ... do something ...
    except aiomqtt.ConnectError as e:
        print(f"Connection failed: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

asyncio.run(main())
```
```

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

### Publish Periodically

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/quick-reference.md

Example of publishing messages at a regular interval using asyncio.Tasks. This pattern is useful for periodic status updates or data logging.

```python
import asyncio

async def publish_loop(client):
    counter = 0
    while True:
        await client.publish("app/counter", str(counter).encode())
        counter += 1
        await asyncio.sleep(5)

async with aiomqtt.Client(hostname="broker.example.com") as client:
    await asyncio.gather(
        client.messages(),
        publish_loop(client),
    )
```

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

### Reconnection Flow Example

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Describes the process of the client automatically reconnecting to the broker after a disconnection, including the use of exponential backoff for retry intervals.

```python
# The reconnection flow is managed by the client's internal logic when configured.
# When a disconnection occurs, the client will attempt to reconnect
# using the specified reconnect_interval and max_reconnect_interval.
# This involves sending a new CONNECT packet.

```

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

### Basic aiomqtt Usage Pattern

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/quick-reference.md

Demonstrates the fundamental pattern for using aiomqtt: connecting to a broker, publishing a message, subscribing to a topic, and receiving messages within an asynchronous context.

```python
import asyncio
import aiomqtt

async def main():
    # Connect to broker
    async with aiomqtt.Client(hostname="test.mosquitto.org") as client:
        
        # Publish a message
        await client.publish("test/topic", b"Hello MQTT")
        
        # Subscribe to topics
        await client.subscribe(aiomqtt.TopicFilter("test/topic"))
        
        # Receive messages
        async for message in client.messages():
            print(f"Received: {message.payload}")

asyncio.run(main())
```

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

### Publish with QoS=2

Source: https://github.com/empicano/aiomqtt/blob/main/docs/guides.md

Demonstrates publishing a message with QoS level EXACTLY_ONCE, which involves a four-part handshake (PUBLISH, PUBREC, PUBREL, PUBCOMP). The code includes logic to handle connection loss and retries for PUBLISH and PUBREL packets.

```python
import asyncio
import aiomqtt


async def main() -> None:
    async with aiomqtt.Client(
        hostname="test.mosquitto.org",
        clean_start=False,
        session_expiry_interval=600,
        reconnect=True,
    ) as client:
        packet_id = next(client.packet_ids)
        duplicate = False
        while True:
            try:
                await client.publish(
                    "ducks/louie/status",
                    b"quack",
                    qos=aiomqtt.QoS.EXACTLY_ONCE,
                    packet_id=packet_id,
                    duplicate=duplicate,
                )
                break
            except aiomqtt.ConnectError:
                duplicate = True
                await client.connected()
        while True:
            try:
                await client.pubrel(packet_id)
                break
            except aiomqtt.ConnectError:
                await client.connected()


asyncio.run(main())
```

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

### Client Constructor

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/api-reference-client.md

Instantiate the aiomqtt Client class. Configure connection details, authentication, session management, and client behavior. Ensure hostname and unix_socket are mutually exclusive.

```python
Client(
    *, 
    hostname: str | None = None,
    port: int = 1883,
    unix_socket: str | None = None,
    identifier: str | None = None,
    logger: logging.Logger | None = None,
    ssl_context: ssl.SSLContext | None = None,
    username: str | None = None,
    password: bytes | None = None,
    clean_start: bool = False,
    session_expiry_interval: int = 0,
    reconnect: bool = False,
    will: Will | None = None,
    keep_alive: int = 0,
    authentication_method: str | None = None,
    authentication_data: bytes | None = None,
    request_problem_info: bool = True,
    request_response_info: bool = False,
    receive_max: int = 65535,
    topic_alias_max: int = 0,
    max_packet_size: int | None = None,
    user_properties: list[tuple[str, str]] | None = None,
) -> None
```

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

### Connect to local MQTT broker

Source: https://github.com/empicano/aiomqtt/blob/main/CONTRIBUTING.md

Initialize the aiomqtt client to connect to a local broker instance.

```python
aiomqtt.Client("localhost")
```

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

### Connect, Publish, and Subscribe with aiomqtt

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/README.md

Connect to a specified MQTT broker, publish a message to a topic, subscribe to that topic, and then print received messages. Ensure you have an asyncio event loop running.

```python
import asyncio
import aiomqtt

async def main():
    # Connect to MQTT broker
    async with aiomqtt.Client(hostname="test.mosquitto.org") as client:
        # Publish a message
        await client.publish("test/topic", b"Hello MQTT")
        
        # Subscribe to topic
        await client.subscribe(aiomqtt.TopicFilter("test/topic"))
        
        # Receive messages
        async for message in client.messages():
            print(f"Received: {message.payload}")

asyncio.run(main())
```

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

### Publish with QoS 0 (Fire and Forget)

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/protocol-flow.md

Use QoS 0 for the fastest publish method when message delivery is not critical. It requires no packet ID and receives no acknowledgment from the broker.

```python
# QoS=0 requires no packet_id
await client.publish(
    "topic/test",
    b"Fire and forget",
    qos=aiomqtt.QoS.AT_MOST_ONCE,
)
# Returns immediately, None returned
```

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

### Use aiomqtt Client as Async Context Manager

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/api-reference-client.md

Establishes a connection to the MQTT broker and manages background tasks. Use this pattern for reliable connection handling and automatic cleanup.

```python
async with aiomqtt.Client(hostname="broker.example.com") as client:
    await client.publish("topic/test", b"Hello MQTT")
    await client.subscribe("topic/test")
    async for message in client.messages():
        print(message.payload)
```

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

### Client Class Documentation

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/MANIFEST.txt

Detailed documentation for the aiomqtt Client class, including its constructor, connection management, publish/subscribe methods, and message handling.

```APIDOC
## Client Class API

### Description
This section details the `Client` class, which is the primary interface for interacting with MQTT brokers using aiomqtt. It covers initialization, connection, message publishing and subscription, and acknowledgment handling.

### Constructor

#### Parameters
- **host** (str) - The MQTT broker hostname or IP address.
- **port** (int) - The MQTT broker port. Defaults to 1883.
- **client_id** (str, optional) - The unique identifier for the MQTT client. If not provided, a random one will be generated.
- **keepalive** (int) - Maximum period in seconds between client and broker communications. Defaults to 60.
- **ssl** (bool) - Whether to use SSL/TLS for the connection. Defaults to False.
- **ssl_params** (dict, optional) - Dictionary of parameters for SSL/TLS context.
- **loop** (asyncio.AbstractEventLoop, optional) - The asyncio event loop to use. If not provided, the default loop is used.
- **clean_session** (bool) - Whether to start a clean session. Defaults to True.
- **protocol_version** (int) - MQTT protocol version. Defaults to 4 (MQTT 3.1.1).
- **will_topic** (str, optional) - The topic for the Last Will and Testament message.
- **will_qos** (int) - The QoS level for the Last Will and Testament message. Defaults to 0.
- **will_retain** (bool) - Whether the Last Will and Testament message should be retained. Defaults to False.
- **will_message** (str, optional) - The payload of the Last Will and Testament message.
- **auth** (tuple, optional) - Authentication credentials as a tuple (username, password).
- **reconnect_retries** (int) - Maximum number of reconnection attempts. Defaults to -1 (infinite).
- **reconnect_delay** (int) - Delay in seconds between reconnection attempts. Defaults to 1.
- **reconnect_max_delay** (int) - Maximum delay in seconds between reconnection attempts. Defaults to 60.
- **ping_interval** (int) - Interval in seconds for sending PINGREQ messages. Defaults to 0 (disabled).
- **ping_timeout** (int) - Timeout in seconds for PINGRESP. Defaults to 15.
- **socket_keepalive** (bool) - Whether to enable TCP keepalive. Defaults to False.
- **socket_connect_timeout** (int) - Timeout in seconds for establishing the socket connection. Defaults to 10.
- **socket_timeout** (int) - Timeout in seconds for socket operations. Defaults to None.
- **max_queued_messages** (int) - Maximum number of messages that can be queued for sending. Defaults to 100.
- **max_retained_messages** (int) - Maximum number of retained messages to store. Defaults to 0 (unlimited).
- **extra_properties** (dict, optional) - Dictionary of MQTT v5.0 properties.
- **session_expiry_interval** (int, optional) - Session expiry interval in seconds for MQTT v5.0.
- **connect_properties** (dict, optional) - Dictionary of MQTT v5.0 CONNECT properties.

### Context Manager Methods

#### `__aenter__()`

- **Description**: Enters the runtime context related to this object. Used with `async with` for automatic connection and disconnection.
- **Returns**: The `Client` instance.

#### `__aexit__(exc_type, exc_val, exc_tb)`

- **Description**: Exits the runtime context. Ensures the client is disconnected gracefully.
- **Parameters**:
  - **exc_type**: Exception type if an exception occurred within the `async with` block.
  - **exc_val**: Exception value.
  - **exc_tb**: Traceback object.

### Connection Status Methods

#### `connected()`

- **Description**: Returns `True` if the client is currently connected to the broker, `False` otherwise.
- **Returns**: `bool`

#### `disconnected()`

- **Description**: Returns `True` if the client is currently disconnected from the broker, `False` otherwise.
- **Returns**: `bool`

### Publish Method

#### `publish(topic, payload=b'', qos=0, retain=False, properties=None)`

- **Description**: Publishes a message to a specified topic.
- **Parameters**:
  - **topic** (str) - The topic to publish the message to.
  - **payload** (bytes or str) - The message payload. Defaults to an empty byte string.
  - **qos** (int) - The Quality of Service level (0, 1, or 2). Defaults to 0.
  - **retain** (bool) - Whether the message should be retained by the broker. Defaults to False.
  - **properties** (dict, optional) - MQTT v5.0 publish properties.
- **Returns**: A `PublishPacket` object upon successful acknowledgment (for QoS 1 and 2).

### Acknowledgment Methods

#### `puback(packet_id)`

- **Description**: Acknowledges a received PUBLISH packet with QoS 1.
- **Parameters**:
  - **packet_id** (int) - The ID of the PUBLISH packet to acknowledge.

#### `pubrec(packet_id)`

- **Description**: Sends a PUBREC packet in response to a received PUBREC packet (QoS 2).
- **Parameters**:
  - **packet_id** (int) - The ID of the PUBREC packet.

#### `pubrel(packet_id)`

- **Description**: Sends a PUBREL packet to acknowledge a received PUBREC packet (QoS 2).
- **Parameters**:
  - **packet_id** (int) - The ID of the PUBREC packet.

#### `pubcomp(packet_id)`

- **Description**: Sends a PUBCOMP packet to acknowledge a received PUBREL packet (QoS 2).
- **Parameters**:
  - **packet_id** (int) - The ID of the PUBREL packet.

### Subscribe and Unsubscribe Methods

#### `subscribe(topics, qos=0)`

- **Description**: Subscribes to one or more topics.
- **Parameters**:
  - **topics** (str or dict) - A single topic string or a dictionary mapping topic strings to QoS levels.
  - **qos** (int) - The default QoS level for topics if a dictionary is not provided. Defaults to 0.
- **Returns**: A `SubAckPacket` object.

#### `unsubscribe(topics)`

- **Description**: Unsubscribes from one or more topics.
- **Parameters**:
  - **topics** (str or list) - A single topic string or a list of topic strings to unsubscribe from.
- **Returns**: An `UnsubAckPacket` object.

### Message Iteration

#### `async for message in client:`

- **Description**: Iterates over incoming messages asynchronously. This is the primary way to receive messages from the broker.
- **Yields**: `Message` objects.

### Packet ID Management

- **Description**: The client automatically manages packet IDs for PUBLISH, SUBSCRIBE, and UNSUBSCRIBE messages. These IDs are unique per connection.

### Identifier Management

- **Description**: The client handles client identification, including the `client_id` and session management.

### Usage Examples

#### Connecting and Publishing
```python
import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await client.publish("aiomqtt/test", "Hello world!")
        print("Published message")

asyncio.run(main())
```

#### Subscribing and Receiving
```python
import asyncio
import aiomqtt

async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("aiomqtt/test")
            async for message in messages:
                print(f"Received message: {message.payload.decode()}")
                break # Only receive one message for this example

asyncio.run(main())
```
```

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

### Connect to broker using context manager

Source: https://github.com/empicano/aiomqtt/blob/main/docs/migration-v2.md

The recommended approach for managing client connections using the async with statement.

```python
import asyncio
import aiomqtt


async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await client.publish("temperature/outside", payload=28.4)


asyncio.run(main())
```

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

### Client Constructor

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/api-reference-client.md

Initializes an aiomqtt Client instance. This constructor manages the connection to an MQTT broker with various configuration options for authentication, session management, and network settings.

```APIDOC
## Client Class Constructor

### Description
Initializes an aiomqtt Client instance. This constructor manages the connection to an MQTT broker with various configuration options for authentication, session management, and network settings.

### Parameters

#### Connection Parameters
- **hostname** (str | None): The broker's hostname or IP address. Must not be set with `unix_socket`.
- **port** (int): The broker's network port. Defaults to 1883.
- **unix_socket** (str | None): Path to the broker's Unix domain socket. Must not be set with `hostname`. Only supported on platforms with Unix socket support.

#### Client Identification & Logging
- **identifier** (str | None): Client identifier. Auto-generated if None (format: `aiomqtt-{random_hex}`). The broker may override this value.
- **logger** (logging.Logger | None): Custom logger. Defaults to `logging.getLogger("aiomqtt")` with WARNING level.

#### Security & Authentication
- **ssl_context** (ssl.SSLContext | None): SSL context for TLS encryption. If None, connection is unencrypted.
- **username** (str | None): Username for authentication.
- **password** (bytes | None): Password for authentication. Can carry any binary credential information.
- **authentication_method** (str | None): Name of the extended authentication method.
- **authentication_data** (bytes | None): Extended authentication data defined by the authentication method.

#### Session Management
- **clean_start** (bool): If True, broker discards existing session and creates new one. If False, broker resumes existing session or creates new if none exists. Defaults to False.
- **session_expiry_interval** (int): Session expiry time in seconds. 0 means session expires immediately after disconnect. Defaults to 0.

#### Connection & Keep-Alive
- **reconnect** (bool): If True, enables automatic reconnection with exponential backoff. Defaults to False.
- **will** (Will | None): Will message to publish if client disconnects unexpectedly.
- **keep_alive** (int): Keep alive interval in seconds. 0 disables keep alive. Broker may override this value. Defaults to 0.

#### Protocol Options (MQTTv5)
- **request_problem_info** (bool): If False, broker must not send reason strings or user properties on packets other than PUBLISH, CONNACK, or DISCONNECT. Defaults to True.
- **request_response_info** (bool): If False, broker must not return response information in CONNACK. If True, broker may return response information. Defaults to False.
- **receive_max** (int): Maximum number of unacknowledged QoS=1 and QoS=2 PUBLISH packets the broker may send to client. Defaults to 65535.
- **topic_alias_max** (int): Maximum number of topic aliases the broker may use when sending PUBLISH packets. 0 means no topic aliases. Defaults to 0.
- **max_packet_size** (int | None): Maximum packet size in bytes to accept. None means no limit beyond protocol limitations.
- **user_properties** (list[tuple[str, str]] | None): Name/value pairs sent with CONNECT packet. Same name can appear multiple times. Order is preserved.

### Raises
- **ValueError**: If both `hostname` and `unix_socket` are set, or if neither are set.
- **OSError**: If `unix_socket` is specified but platform does not support Unix domain sockets.
```

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

### Handle Messages Concurrently with asyncio Tasks

Source: https://github.com/empicano/aiomqtt/blob/main/docs/guides.md

Process incoming MQTT messages concurrently using multiple asyncio tasks to avoid bottlenecks. This example simulates I/O-bound work for each message.

```python
import asyncio
import aiomqtt
import random


async def consume(client: aiomqtt.Client) -> None:
    async for message in client.messages():
        await asyncio.sleep(random.random())  # Simulate some I/O-bound work
        print(message.payload)


async def main() -> None:
    async with aiomqtt.Client("test.mosquitto.org", receive_max=16) as client:
        await client.subscribe(aiomqtt.TopicFilter("ducks/#", max_qos=aiomqtt.QoS.AT_MOST_ONCE))
        async with asyncio.TaskGroup() as tg:
            for _ in range(4):  # The number of consumer tasks
                tg.create_task(consume(client))
```

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

### Manually manage client lifecycle

Source: https://github.com/empicano/aiomqtt/blob/main/docs/migration-v2.md

Alternative approach for scenarios where a context manager cannot be used, utilizing __aenter__ and __aexit__ directly.

```python
import asyncio
import aiomqtt


async def main():
    client = aiomqtt.Client("test.mosquitto.org")
    await client.__aenter__()
    try:
        await client.publish("temperature/outside", payload=28.4)
    finally:
        await client.__aexit__(None, None, None)


asyncio.run(main())
```

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

### Subscribe with Topic Wildcards

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/quick-reference.md

Demonstrates subscribing to topics using single-level (+) and multi-level (#) wildcards, as well as exact topic matching.

```python
await client.subscribe(aiomqtt.TopicFilter("home/+/temperature"))
```

```python
await client.subscribe(aiomqtt.TopicFilter("sensors/#"))
```

```python
await client.subscribe(aiomqtt.TopicFilter("home/temperature"))
```

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

### Session Persistence Configuration

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/configuration.md

Configure the client to resume a previous session if available, or create a new one. Set `clean_start=False` to enable persistence and `session_expiry_interval` for session duration.

```python
import aiomqtt

# Resume session if available, otherwise create new
async with aiomqtt.Client(
    hostname="broker.example.com",
    identifier="persistent-client",
    clean_start=False,  # Resume session if available
    session_expiry_interval=3600,  # Session lasts 1 hour
) as client:
    await client.subscribe(aiomqtt.TopicFilter("test/topic"))
```

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

### Configure Authentication with aiomqtt

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/quick-reference.md

Connect to a broker requiring authentication by providing username and password credentials.

```python
async with aiomqtt.Client(
    hostname="broker.example.com",
    username="user",
    password=b"password",
) as client:
    pass
```

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

### Implement Request-Response Pattern

Source: https://github.com/empicano/aiomqtt/blob/main/_autodocs/quick-reference.md

Demonstrates the request-response pattern where a publisher sends a request with correlation data and a responder replies to the specified response topic, using correlation data to match requests and responses.

```python
# Publisher sends request
packet_id = next(client.packet_ids)
await client.publish(
    "request/get_status",
    b"",
    qos=aiomqtt.QoS.AT_LEAST_ONCE,
    packet_id=packet_id,
    response_topic="response/status",
    correlation_data=str(packet_id).encode(),
)
```

```python
# Responder receives request and publishes response
async for message in client.messages():
    if message.topic == "request/get_status":
        response_topic = message.response_topic or "response/status"
        await client.publish(
            response_topic,
            b"OK",
            correlation_data=message.correlation_data,
        )
```