c - 为什么使用 paho.mqtt.c 发送此消息会导致段错误？

Question

我正在尝试使用 MQTT 协议(protocol)发送特定的消息类型。我正在使用 paho.mqtt.c 库，我的代理是 RabbitMQ 3.6.12，运行 Erlang 20.0。我正在运行 CentOS 6.9 的虚拟机上工作。我首先尝试通过为我的特定消息类型创建一个结构体来实现此目的，然后在提出这个问题之前，我还尝试使用 JSON 来创建我的特定消息类型。我安装了 cJSON(来自 here )。

这是我使用 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，所以我会继续调查，以防我使用错误，但任何帮助将不胜感激。我尝试按照评论中的建议缩小范围，但我不知道我能做什么，因为它一旦尝试访问消息的有效负载就会崩溃。

请记住，我只是一名在计算机科学方面经验不超过一年的学生。另外，英语不是我的母语，我希望我能足够清楚地解释自己。

Answer 1

问题已解决(有点)

我只是将整个消息帧作为字符串发送并在 msgarrvd 函数中解析它:

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"

volatile MQTTClient_deliveryToken deliveredtoken;
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int main(int argc, char* argv[])
{   

    char* test = "{\"string\" : \"whatever\", \"entier\" : 42}";

    MQTTClient publisher;
    MQTTClient_connectOptions connexion = MQTTClient_connectOptions_initializer;

    MQTTClient_message msg = msg_creation(test,strlen(test),0,0);

    int rc;
    MQTTClient_create(&publisher, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    connexion.cleansession = 1;
    MQTTClient_setCallbacks(publisher, NULL, connlost, frame_test_arrvd, NULL);
    if ((rc = MQTTClient_connect(publisher,&connexion)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
    }
    MQTTClient_publishMessage(publisher, TOPIC,&msg,&token);
    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 <time.h>
#include <sys/time.h>
#include <cjson/cJSON.h>
#include "MQTTClient.h"

// message frames

void print_frame1(cJSON* frame1)
{
    char * str = cJSON_Print(frame1);
    printf("%s\n",str);
}

int frame_test_arrvd(void* context, char* topicName, int topicLen, MQTTClient_message* msg)
{
    printf("message original : %s\n",msg->payload);
    cJSON* payload_ptr = cJSON_Parse(msg->payload);

    print_frame1(payload_ptr);
    int i = cJSON_GetArraySize(payload_ptr);
    printf("number of items in frame : %d\n",i);
    cJSON* entier = NULL;
    entier = cJSON_GetObjectItem(payload_ptr,"entier");
    char* pt = cJSON_Print(entier);
    printf("entier : %s\n",pt);
    cJSON* str = NULL;
    str = cJSON_GetObjectItem(payload_ptr,"string");
    char* st = cJSON_Print(str);
    printf("string : %s\n",st);

    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;
}

子框架.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_test_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;
}

这并不理想，因为我仍然想创建特定类型的消息，而且我还没有找到导致段错误的原因，但这种方式可以完美地工作。

更新

我找到了一种方法来创建特定类型的消息，并仍然将它们作为字符串发送以在 msgarrvd 函数中解析它们，我刚刚创建了一个函数 cJSON_ToString 将 cJSON* 对象转换为字符串，以便我可以创建一个 cJOSN从我想要的任何结构中获取，然后将其转换为字符串来发送。