时间戳

首页 > 解决方案 > 如何用蚊子测试 python paho mqtt？

python - 如何用蚊子测试 python paho mqtt？

问题描述

我想测试 mosquitto MQTT Python 客户端端口。

import json

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    """Called when connected to MQTT broker."""
    client.subscribe("hermes/intent/#")
    client.subscribe("hermes/nlu/intentNotRecognized")
    print("Connected. Waiting for intents.")

def on_disconnect(client, userdata, flags, rc):
    """Called when disconnected from MQTT broker."""
    client.reconnect()

def on_message(client, userdata, msg):
    """Called each time a message is received on a subscribed topic."""
    nlu_payload = json.loads(msg.payload)
    if msg.topic == "hermes/nlu/intentNotRecognized":
        sentence = "Unrecognized command!"
        print("Recognition failure")
    else:
        # Intent
        print("Got intent:", nlu_payload["intent"]["intentName"])

        # Speak the text from the intent
        sentence = nlu_payload["input"]

    site_id = nlu_payload["siteId"]
    client.publish("hermes/tts/say", json.dumps({"text": sentence, "siteId": site_id}))


# Create MQTT client and connect to broker
client = mqtt.Client()
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message

client.connect("localhost", 1883)
client.loop_forever()

我用命令运行它

$ python script.py` 
Connected. Waiting for intents.

mosquitto 会发送 POST 请求吗？还是我必须向 mosquitto 发起请求？我应该如何创建一个请求以便我得到

Got intent: SetTimer

标签： pythonmqttmosquittopaho

解决方案

MQTT 不是 HTTP，POST 是一个 HTTP 动词，在 MQTT 上下文中没有任何意义。

MQTT 是一个发布/订阅协议，而 HTTP 是一个请求/响应协议。

您发布的代码仅订阅 2 个主题，它不发布任何内容（直到收到消息）。因此，除非您有另一个应用程序将消息发布到 2 个主题中的 1 个主题，否则 python 代码已订阅它只会坐在那里等待消息。

如果需要，您可以使用 mosquitto 命令行工具发送消息。例如

mosquitto_pub -t hermes/intent/foo -m '{"intent": { "intentName": "SetTimer"}}'

推荐阅读