Skip to main content

​
Overview

LoadForge supports MQTT protocol testing via the MqttUser from locust-plugins. You can simulate publishing and subscribing to topics under load, ensuring your broker handles high throughput.

​
Locust Test Script (locust.py)

# locust.py
import time
from locust import between, task
from locust_plugins.users import MqttUser

class MyMqttUser(MqttUser):
    wait_time = between(1, 3)
    host = "mqtt://broker.hivemq.com:1883"

    @task
    def publish_message(self):
        # Publish a message to a topic
        self.client.publish(topic="loadforge/test", payload="Hello from LoadForge")

    @task
    def subscribe_topic(self):
        # Subscribe to a topic
        self.client.subscribe(topic="loadforge/test")
        # Wait for incoming messages
        time.sleep(1)
Notes:
  • Install dependencies: pip install locust locust-plugins paho-mqtt if you wish to test locally.
  • Ensure your MQTT broker is accessible at the specified host.
⌘I