gmqtt

To run the unit tests for gmqtt, you must first install the package with the [test] extra dependencies.

Note: The tests require a flespi.io account. You must provide your flespi.io token via the USERNAME environment variable.

# Install test dependencies
pip3 install .[test]

# Set your flespi.io token
export USERNAME=YOUR_FLESPI_IO_TOKEN

# Run tests
pytest-3 tests

Install the latest stable version of the gmqtt Python async MQTT client using pip:

pip3 install gmqtt

To contribute to gmqtt, it is recommended to use a Python virtual environment. Follow these steps to clone your fork, set up the environment, and install the package in development mode:

1. Clone your fork of the repository.
2. Create and activate a Python virtual environment.
3. Run setup.py develop to install the package for development.

To stay up-to-date with the main project, add the upstream repository as a remote and rebase your local branch periodically.

git clone git@github.com:[YOUR_GITHUB_USERNAME]/gmqtt.git
cd gmqtt
python3 -m venv .
source bin/activate
python3 setup.py develop

# To sync with upstream
git remote add upstream git@github.com:wialon/gmqtt.git
git pull --rebase upstream master

When initializing the MQTTClient, you can pass MQTT 5.0 connection properties as keyword arguments. These are stored in client.properties after connection.

• session_expiry_interval (int): Session expiry in seconds. 0 or absent means the session ends when the network connection closes. 0xFFFFFFFF means the session never expires.
• receive_maximum (int): Limits the number of QoS 1 and QoS 2 publications processed concurrently.
• user_property (tuple(str, str)): Key-value pairs for diagnostic information.
• maximum_packet_size (int): Informs the server of the maximum packet size (in bytes) the client will process.

client = gmqtt.Client("lenkaklient", receive_maximum=24000, session_expiry_interval=60, user_property=('myid', '12345'))

In MQTT 5.0, the session_expiry_interval determines how long the broker maintains the session after the client disconnects.

• Value 0 (or absent): The session ends when the network connection is closed.
• Value 0xFFFFFFFF: The session never expires.
• Server Override: The broker may override your requested value in the CONNACK packet. You can check the effective interval using the client.session_expiry_interval property after connecting.

By default, the client attempts to reconnect indefinitely. You can customize the number of retries and the delay between attempts using set_config.

• reconnect_retries (int): Number of attempts. Default is -1 (infinity).
• reconnect_delay (int): Delay in seconds. Default is 6.

client = MQTTClient("client-id")
client.set_config({'reconnect_retries': 10, 'reconnect_delay': 60})

You can define on_message as an async function. To signal successful processing, the callback must return a valid PUBACK code (e.g., 0 for success).

async def on_message(client, topic, payload, qos, properties):
    # Process message
    pass
    return 0

gmqtt uses MQTT version 5.0 by default. If the broker does not support 5.0, the client attempts to downgrade to 3.1.1 and reconnect automatically. You can explicitly force a specific version in the connect method using constants from gmqtt.mqtt.constants.

from gmqtt.mqtt.constants import MQTTv311
client = MQTTClient('clientid')
client.set_auth_credentials(token, None)
await client.connect(broker_host, 1883, keepalive=60, version=MQTTv311)

When publishing messages, you can include MQTT 5.0 properties. These properties are sent in the publish packet and will be passed to the on_message callback on the receiving end.

• message_expiry_interval (int): Lifetime of the application message in seconds.
• content_type (unicode): UTF-8 string describing the content of the message.
• user_property (tuple(str, str)): Key-value pairs.
• subscription_identifier (int): Sent by the broker.
• topic_alias (int): Allows publishing with an empty string topic after an initial message with a full topic string and a topic_alias.

# Example publishing with properties
client.publish('TEST/TIME', str(time.time()), qos=1, retain=True, message_expiry_interval=60, content_type='json')

The Client exposes several properties that allow you to assign callback functions for MQTT events. These callbacks are triggered by the internal MqttPackageHandler.

Note: The on_message callback receives the payload as bytes.

async def on_message(client, topic, payload, qos, properties):
    print(f"Received {payload} on {topic}")

client.on_message = on_message