c - 使用 MQTT paho 异步发布触发新消息回调
问题描述
我正在使用来自 MQTT paho 的 MQTTAsync_publish.c 示例代码。我所做的是将示例代码中的常量 PAYLOAD 替换为 SendMessageArray。如果有新的无线电消息到达,这个数组的内容会在我的其他代码中发生变化。有效负载在第一次连接时成功发送。
但是,我不明白当我的有效负载更改时如何触发新发布?
函数 MQTTAsync_sendMessage 是在 onConnect 回调中触发的,所以我必须以某种方式再次触发此回调,但是如何。如果我通过不调用 onSend 回调将示例代码更改为在第一次发送后保持连接,那么在完成 = 0 时,它是“等待客户端 ID:%s\n 的主题 %s 上的 %s 的发布”,但我无法放置新的有效载荷。
volatile MQTTAsync_token deliveredtoken;
char SendMessageArray[4096] = { 0 }; // global
int finished = 0;
void connlost(void *context, char *cause)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
printf("\nConnection lost\n");
printf(" cause: %s\n", cause);
printf("Reconnecting\n");
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
finished = 1;
}
}
void onDisconnect(void* context, MQTTAsync_successData* response)
{
printf("Successful disconnection\n");
finished = 1;
}
void onSend(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
int rc;
printf("Message with token value %d delivery confirmed\n", response->token);
opts.onSuccess = onDisconnect;
opts.context = client;
if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start sendMessage, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
printf("Connect failed, rc %d\n", response ? response->code : 0);
finished = 1;
}
void onConnect(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
int rc;
printf("Successful connection\n");
opts.onSuccess = onSend; // -> something like onHoldConnect???
opts.context = client;
pubmsg.payload = SendMessageArray; //PAYLOAD;
pubmsg.payloadlen = (int)strlen(SendMessageArray);
pubmsg.qos = QOS;
pubmsg.retained = 0;
deliveredtoken = 0;
if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start sendMessage, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
int main(int argc, char* argv[])
{
MQTTAsync client;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
MQTTAsync_setCallbacks(client, NULL, connlost, NULL, NULL);
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
conn_opts.onSuccess = onConnect;
conn_opts.onFailure = onConnectFailure;
conn_opts.context = client;
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
exit(EXIT_FAILURE);
}
printf("Waiting for publication of %s\n"
"on topic %s for client with ClientID: %s\n",
SendMessageArray, TOPIC, CLIENTID);
while (!finished)
#if defined(WIN32)
Sleep(100);
#else
usleep(10000L);
#endif
MQTTAsync_destroy(&client);
return rc;
}
这里是原始代码: http: //gitlab.witium.com.cn/eclipse/paho.mqtt.c/blob/master/src/samples/MQTTAsync_publish.c
我不知何故需要与发送所有新消息的代理建立持久连接。谢谢你的帮助。
编辑:好的,我现在所做的是(1)通过在发送后不断开连接来保持连接:
void onHoldConnect(void* context, MQTTAsync_successData* response)
{
printf("Holding connection\n");
}
void onConnect(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
int rc;
printf("Successful connection\n");
opts.onSuccess = onHoldConnect; //onSend;
opts.onFailure = onSendFailure;
opts.context = client;
pubmsg.payload = SendMessageArray; //PAYLOAD;
pubmsg.payloadlen = (int)strlen(SendMessageArray);
pubmsg.qos = QOS;
pubmsg.retained = 0;
if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start sendMessage, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
(2) 如果 SendMessageArray 已被修改,则在 while 循环中使用全局 bool 运行 onConnect 函数:
while (!finished) {
if (sendFlag == true) {
sendFlag = false;
onConnect(client, &conn_opts);
}
#if defined(_WIN32)
Sleep(100);
#else
usleep(10000L);
#endif
}
这是解决问题的正确方法吗?我只是再次使用 (client, &conn_opts) 作为参数。
解决方案
推荐阅读
- flutter - 如何在 Flutter 应用程序中将事件添加到日历?
- powerbi - 如何从 bitbucket 查看 git repo 中 Power BI 文件的差异?
- reactjs - 可以使用反应状态来扩展特定路由器链接负载时的应用栏高度吗?
- node.js - 如何在 Watson Assistant 中设置多个 Webhook
- c++ - 避免在 PPTP 连接断开时提示用户重拨
- powershell - 如何使用PowerShell将文件提取到路径中包含点字符的特定位置?
- html - 我的页脚的一大块从页面上消失了
- c# - 网站刷新后保存随机数值
- javascript - 如何为具有相同 JS 文件的不同页面指定代码
- json - 将动态 json 元素解析为同一字段