我正在尝试使用 MQTT 协议发送特定的消息类型。我正在使用 paho.mqtt.c 库，我的代理是 RabbitMQ 3.6.12，运行 Erlang 20.0。我正在使用运行 CentOS 6.9 的虚拟机。我首先尝试通过为我的特定消息类型创建一个结构来做到这一点，然后在问这个问题之前，我还尝试使用 JSON 来创建我的特定消息类型。我安装了 cJSON（从这里）。
这是我使用 cJSON 的整个代码：
pubframe.c：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "../tools.c"
#include <cjson/cJSON.h>
#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "MY_PUB"
#define TOPIC "MQTT/Test"
int main(int argc, char* argv[])
{
frame1 test = {42,"test"};
cJSON* frm = NULL;
frm = cJSON_CreateObject();
cJSON_AddNumberToObject(frm,"entier",test.E);
cJSON_AddStringToObject(frm,"string",test.S);
print_frame1(frm);
int i = cJSON_GetArraySize(frm);
printf("number of items in frame : %d\n",i);
cJSON* entier = NULL;
entier = cJSON_GetObjectItem(frm,"entier");
char* pt = cJSON_Print(entier);
printf("entier : %s\n",pt);
cJSON* str = NULL;
str = cJSON_GetObjectItem(frm,"string");
char* st = cJSON_Print(str);
printf("string : %s\n",st);
printf("size of message : %d\n",sizeof(cJSON));
MQTTClient publisher;
MQTTClient_connectOptions connexion = MQTTClient_connectOptions_initializer;
MQTTClient_message msg = msg_creation(frm,sizeof(cJSON),0,0);
MQTTClient_deliveryToken token;
int rc;
MQTTClient_create(&publisher, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
connexion.cleansession = 1;
MQTTClient_setCallbacks(publisher, NULL, connlost, frame_json_arrvd, NULL);
if ((rc = MQTTClient_connect(publisher,&connexion)) != MQTTCLIENT_SUCCESS)
{
printf("Failed to connect, return code %d\n", rc);
}
MQTTClient_publishMessage(publisher, TOPIC,&msg,NULL);
printf("Message sent!\n");
cJSON_Delete(frm);
MQTTClient_disconnect(publisher,10000);
MQTTClient_destroy(&publisher);
return rc;
}
子框架.c：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "../tools.c"
#include <cjson/cJSON.h>
#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ClientID"
#define TOPIC "MQTT/Test"
#define QOS 0
#define TIMEOUT 10000L
int main(int argc, char* argv[])
{
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
int rc;
int ch;
MQTTClient_create(&client, ADDRESS, CLIENTID,
MQTTCLIENT_PERSISTENCE_NONE, NULL);
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
MQTTClient_setCallbacks(client, NULL, connlost, frame_json_arrvd, NULL);
if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
{
printf("Failed to connect, return code %d\n", rc);
exit(EXIT_FAILURE);
}
printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
"Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
MQTTClient_subscribe(client, TOPIC, QOS);
do
{
ch = getchar();
} while(ch!='Q' && ch != 'q');
MQTTClient_disconnect(client, 10000);
MQTTClient_destroy(&client);
return rc;
}
工具.c：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>
#include <cjson/cJSON.h>
#include "MQTTClient.h"
int compteur;
// message frames
typedef struct frame1
{
int E;
char* S;
} frame1;
void print_frame1bis(frame1* F)
{
printf(" %s\n",F->S);
printf(" %d\n",F->E);
}
void print_frame1(cJSON* frame1)
{
char * str = cJSON_Print(frame1);
printf("%s\n",str);
}
int frame_json_arrvd(void* context, char* topicName, int topicLen, MQTTClient_message* msg)
{
cJSON* payload_ptr = NULL;
printf("size of message : %d\n",sizeof(msg->payload));
int j = cJSON_GetArraySize(msg->payload);
printf("ok\n");
printf("number of items in frame : %d\n",j);
payload_ptr = cJSON_CreateObject();
payload_ptr = msg->payload;
int i = cJSON_GetArraySize(payload_ptr);
printf("number of items in frame : %d\n",i);
print_frame1(payload_ptr);
cJSON_Delete(payload_ptr);
MQTTClient_freeMessage(&msg);
MQTTClient_free(topicName);
return 1;
}
// in case connexion is lost
void connlost(void *context, char *cause)
{
printf("\nConnection lost\n");
printf(" cause: %s\n", cause);
}
// create message
MQTTClient_message msg_creation(void* payload, int length, int qos, int retained)
{
MQTTClient_message pubmsg = MQTTClient_message_initializer;
pubmsg.payload = payload;
pubmsg.payloadlen = length;
pubmsg.qos = qos;
pubmsg.retained = retained;
return pubmsg;
}
订阅者的输出是
Subscribing to topic MQTT/Test
for client ClientID using QoS0
Press Q<Enter> to quit
size of message : 8
Segmentation fault (core dumped)
因此，似乎没有正确接收到消息，因为订阅者在尝试获取项目数时立即崩溃。
这是我运行 gdb 时得到的，不确定是否有帮助。
Program terminated with signal 11, Segmentation fault.
#0 0x00007f4e49a8b63e in cJSON_GetArraySize () from /usr/lib64/libcjson.so.1
(gdb) bt
#0 0x00007f4e49a8b63e in cJSON_GetArraySize () from /usr/lib64/libcjson.so.1
#1 0x00000000004010c9 in frame_json_arrvd (context=0x0,
topicName=0x7f4e440009e4 "MQTT/Test", topicLen=0, msg=0x7f4e44000bb4)
at ../tools.c:110
#2 0x00007f4e49c962a5 in MQTTClient_run (n=<value optimized out>)
at src/MQTTClient.c:604
#3 0x000000378b807aa1 in start_thread (arg=0x7f4e49a83700) at pthread_create.c:301
#4 0x000000378b4e8bcd in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:115
发布者的输出是
{
"entier": 42,
"string": "test"
}
number of items in frame : 2
entier : 42
string : "test"
size of message : 64
Message sent!
所以看起来消息是正确创建的。
我最近刚刚研究了 cJSON，所以我会继续调查以防我使用错了，但我们将不胜感激。我试图按照评论中的建议缩小范围，但我看不到我能做什么，因为它一旦尝试访问消息的有效负载就会崩溃。
请记住，我只是一名在计算机科学领域拥有不超过一年经验的学生。另外，英语不是我的母语，我希望我能清楚地解释自己。