Using MQTT for Inter-Process Communication

Introduction

The previous article shared: “A Simple Understanding of MQTT and Mosquitto”, which only covered some concepts of MQTT and an introduction to Mosquitto. Then readers urged for more:

In this article, we will share an example of using MQTT for inter-process communication. We will modify the comprehensive demo from the previous article “Examples of Practical Knowledge in Embedded Systems: JSON Data”. The functionality of that comprehensive demo is as follows:

This uses sockets for inter-process communication, and this demo is written based on Windows, requiring specific Windows header files.

In this note, we will modify the above comprehensive demo to:

We will use MQTT for inter-process communication, testing the processes under Linux.

First, let’s show the code:

json_print Process Source Code

“json_print.c:”

Swipe to view all code>>>

/*
- Program Function: Package JSON format data and send (MQTT Publisher Client Program)
- Compile Command: gcc cJSON.c json_print.c -L ../mosquitto/build/lib -lmosquitto -o json_print
- Export Mosquitto Dynamic Library: export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH
- Author: ZhengN
- WeChat Official Account: Embedded Mixed Bag
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "cJSON.h"
#include "../mosquitto/lib/mosquitto.h"

#define STU_NAME_LEN 32

/* Student Structure */
typedef struct _Student
{
    char name[STU_NAME_LEN];  // Name
    int num;                  // Student Number
    int c_score;              // C Language Score
} StudentDef, *pStudentDef;

/* Internal Function Declarations */
static StudentDef StudentData_Prepare(void);
static char *StudentsData_Packet(pStudentDef _Stu);
static void StudentData_Send(const char *_data);
static void MqttClientInit(void);
static void MqttClientClean(void);

bool clean_session = true;
struct mosquitto *mosq = NULL;

/********************************************************************************************************
** Function: main
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Description: 
** Returns: 
********************************************************************************************************/
int main(void)
{
    StudentDef stu = {0};
    char *stu_data = NULL;
    int stu_count = 0;
    int i = 0;

    /* MQTT Client Initialization */
    MqttClientInit();

    /* Total number of students to register */
    printf("Please input number of student: ");
    scanf("%d", &stu_count);

    while (i++ < stu_count)
    {
        /* Prepare data */
        stu = StudentData_Prepare();

        /* Package JSON format data */
        stu_data = StudentsData_Packet(&stu);

        /* Send data */
        StudentData_Send(stu_data);
    }

    /* Cleanup operations */
    MqttClientClean();

    return 0;
}

/********************************************************************************************************
** Function: StudentData_Prepare, Prepare data needed for packaging
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Description: 
** Returns: Retrieved data
********************************************************************************************************/
static StudentDef StudentData_Prepare(void)
{
    char name[STU_NAME_LEN] = {0};
    int num = 0;
    int c_score = 0;
    StudentDef stu;

    /* Name */
    printf("Please input name: ");
    scanf("%s", name);
    if (strlen(name) < STU_NAME_LEN)
    {
        strncpy((char*)&stu.name, name, strlen(name)+1);
    }
    else
    {
        printf("The name is too long\n");
    }
    
    /* Student Number */
    printf("Please input num (0~100): ");
    scanf("%d", &num);
    stu.num = num;

    /* C Language Score */
    printf("Please input c_score (0~100): ");
    scanf("%d", &c_score);
    stu.c_score = c_score;

    return stu;
}

/********************************************************************************************************
** Function: StudentsData_Packet, Package JSON format data
**------------------------------------------------------------------------------------------------------
** Parameters: _Stu: Data needed to package student json data
** Description: 
** Returns: JSON formatted string 
********************************************************************************************************/
static char *StudentsData_Packet(pStudentDef _Stu)
{
    char *res_string = NULL;    // Return value
    cJSON *name = NULL;         // Name
    cJSON *num = NULL;          // Student Number
    cJSON *c_score = NULL;      // C Language Score

    /* Create a JSON object, enclosed in {} */
    cJSON *obj = cJSON_CreateObject();
    if (obj == NULL)
    {
        goto end;
    }

    /* Create "name": "xxx" key-value pair */
    name = cJSON_CreateString(_Stu->name);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "name", name);

    /* Create "num": 207 key-value pair */
    num = cJSON_CreateNumber(_Stu->num);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "num", num);
    
    /* Create "c_score": 95 key-value pair */
    c_score = cJSON_CreateNumber(_Stu->c_score);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "c_score", c_score); 

    res_string = cJSON_Print(obj);          // Present in JSON format 
    // res_string = cJSON_PrintUnformatted(obj);   // Present in unformatted form

    if (res_string == NULL)
    {
        fprintf(stderr, "Failed to print monitor.\n");
    }

/* Exception handling: delete (free) */
end:
    cJSON_Delete(obj);
    return res_string;
}

/********************************************************************************************************
** Function: StudentData_Send, Package and send JSON formatted string data
**------------------------------------------------------------------------------------------------------
** Parameters: _data: Data to be sent
** Description: 
** Returns: JSON formatted string 
********************************************************************************************************/
static void StudentData_Send(const char *_data)
{
    printf("%s: %s\n\n", __FUNCTION__, _data);

    /* Publish message */
    mosquitto_publish(mosq, NULL, "test_topic", strlen(_data)+1, (const char*)_data, 0, 0);
}

/********************************************************************************************************
** Function: MqttClientInit, MQTT Client Initialization
**------------------------------------------------------------------------------------------------------
** Parameters: void
** Description: 
** Returns: 
********************************************************************************************************/
static void MqttClientInit(void)
{
    /* Initialize libmosquitto library */
    mosquitto_lib_init();

    /* Create mosquitto client */
    mosq = mosquitto_new(NULL, clean_session, NULL);
    if(NULL == mosq)
    {
        printf("Create mqtt client failed...\n");
        mosquitto_lib_cleanup();
        return;
    }

    /* Connect to server */
    if(mosquitto_connect(mosq, "localhost", 1883, 60))
    {
        printf("Unable to connect...\n");
        return;
    }

    /* Network message processing thread */
    int loop = mosquitto_loop_start(mosq);
    if(loop != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto loop error\n");
        return;
    }
}

/********************************************************************************************************
** Function: MqttClientClean, MQTT Client Cleanup Operations
**------------------------------------------------------------------------------------------------------
** Parameters: void
** Description: 
** Returns: 
********************************************************************************************************/
static void MqttClientClean(void)
{
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
}

json_parse Process Source Code

“json_parse.c:”

Swipe to view all code>>>

/*
- Program Function: Receive JSON data and parse (MQTT Subscriber Client Program)
- Compile Command: gcc cJSON.c json_parse.c -L ../mosquitto/build/lib -lmosquitto -o json_parse
- Export Mosquitto Dynamic Library: export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH
- Author: ZhengN
- WeChat Official Account: Embedded Mixed Bag
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "cJSON.h"
#include "../mosquitto/lib/mosquitto.h"

#define STU_NAME_LEN 32

/* Student Structure */
typedef struct _Student
{
    char name[STU_NAME_LEN];  // Name
    int num;                  // Student Number
    int c_score;              // C Language Score
} StudentDef, *pStudentDef;

/* Internal Function Declarations */
static void StudentsData_Parse(pStudentDef _Stu, const char *_JsonStudnetData);
static void PrintParseResult(const pStudentDef _Stu);
static void SaveParseResult(const pStudentDef _Stu);
static void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message);
static void my_connect_callback(struct mosquitto *mosq, void *userdata, int result);

/* Internal Global Variables */
static FILE *stu_fp = NULL;

/********************************************************************************************************
** Function: main
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Description: 
** Returns: 
********************************************************************************************************/
bool clean_session = true;
int main(void)
{         
    struct mosquitto *mosq = NULL;

    /* Initialize libmosquitto library */
    mosquitto_lib_init();

    /* Create mosquitto client */
    mosq = mosquitto_new(NULL, clean_session, NULL);
    if(NULL == mosq)
    {
        printf("Create mqtt client failed...\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    /* Bind connection and message receiving callback functions */
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_message_callback_set(mosq, my_message_callback);

    /* Connect to server */
    if(mosquitto_connect(mosq, "localhost", 1883, 60))
    {
        printf("Unable to connect...\n");
        return 1;
    }

    /* Loop to process network messages */
    mosquitto_loop_forever(mosq, -1, 1);

    /* Cleanup operations */
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();

    return 0;
}

/********************************************************************************************************
** Function: StudentsData_Parse, Parse JSON format student final data
**------------------------------------------------------------------------------------------------------
** Parameters: _JsonStudnetData: JSON data   _Stu: Save useful parsed data
** Description: 
** Returns: 
********************************************************************************************************/
static void StudentsData_Parse(pStudentDef _Stu, const char *_JsonStudnetData)
{
    cJSON *student_json = NULL;   // student_json operation object, can represent content enclosed in {}
    cJSON *name = NULL;             
    cJSON *num = NULL;
    cJSON *c_score = NULL;
    
    /* Start parsing */
    student_json = cJSON_Parse(_JsonStudnetData);
    if (NULL == student_json)
    {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL)
        {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        goto end;
    }

    /* Parse to get the value of name */
    name = cJSON_GetObjectItemCaseSensitive(student_json, "name");
    if (cJSON_IsString(name) && (name->valuestring != NULL))
    {
        memset(&_Stu->name, 0, STU_NAME_LEN*sizeof(char));
        memcpy(&_Stu->name, name->valuestring, strlen(name->valuestring));
    }

    /* Parse to get the value of num */
    num = cJSON_GetObjectItemCaseSensitive(student_json, "num");
    if (cJSON_IsNumber(num))
    {
        _Stu->num = num->valueint;
    }

    /* Parse to get the value of c_score */
    c_score = cJSON_GetObjectItemCaseSensitive(student_json, "c_score");
    if (cJSON_IsNumber(c_score))
    {
        _Stu->c_score = c_score->valueint;
    }

end:
    cJSON_Delete(student_json);
}

/********************************************************************************************************
** Function: PrintParseResult, Print output parse results
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Description: 
** Returns: 
********************************************************************************************************/
static void PrintParseResult(const pStudentDef _Stu)
{
    printf("name: %s, num: %d, c_score: %d\n\n", _Stu->name, _Stu->num, _Stu->c_score);
}

/********************************************************************************************************
** Function: SaveParseResult, Save parse results
**------------------------------------------------------------------------------------------------------
** Parameters: _Stu: Data to be saved
** Description: 
** Returns: 
********************************************************************************************************/
static void SaveParseResult(const pStudentDef _Stu)
{
    char write_buf[512] = {0};
    static int stu_count = 0;

    /* Open file in a way that allows appending content to the end of the file */
 if((stu_fp = fopen("ParseResult.txt", "a+")) == NULL)
 {
  printf("Open file error!\n");
  return exit(EXIT_FAILURE);
 } 

    /* Write to file in specified format */
snprintf(write_buf, 512, "name: %s, num: %d, c_score: %d\n", _Stu->name, _Stu->num, _Stu->c_score);
    size_t len = fwrite((char*)write_buf, 1, strlen(write_buf), stu_fp);

    /* File position pointer offset */
fseek(stu_fp, len * stu_count, SEEK_SET);
    stu_count++;

    /* Close file */
fclose(stu_fp);
}

/********************************************************************************************************
** Function: my_message_callback, Message receiving callback function
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Returns: 
********************************************************************************************************/
static void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
    StudentDef stu = {0};

    if(message->payloadlen)
    {
        printf("%s %s\n", message->topic, (char*)message->payload);

        /* Parse JSON data */
        StudentsData_Parse(&stu, (const char*)message->payload);  

        /* Print output parse results */
        PrintParseResult(&stu);  

        /* Save data to file */
        SaveParseResult(&stu); 
    }
    else
    {
        printf("%s (null)\n", message->topic);
    }
    fflush(stdout);
}

/********************************************************************************************************
** Function: my_connect_callback, Connection callback function
**------------------------------------------------------------------------------------------------------
** Parameters: 
** Returns: 
********************************************************************************************************/
static void my_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    if(!result)
    {
        /* Subscribe to messages on the test_topic topic */
        mosquitto_subscribe(mosq, NULL, "test_topic", 0);
    }
    else
    {
        fprintf(stderr, "Connect failed\n");
    }
}

Compilation and Execution

1. Compile to generate json_parse and json_print programs:

Swipe to view all code>>>

gcc cJSON.c json_parse.c -L ../mosquitto/build/lib -lmosquitto -o json_parse
gcc cJSON.c json_print.c -L ../mosquitto/build/lib -lmosquitto -o json_print    

Here, the executable programs are generated using the dynamic library linking method. For information on dynamic and static linking, please refer to previous notes: “Supplement on Static and Dynamic Linking (Linux)”, “What are Dynamic and Static Linking?”.

2. Execute json_parse and json_print programs:

Executing these two programs will result in errors:

./json_parse: error while loading shared libraries: libmosquitto.so.1: cannot open shared object file: No such file or directory
./json_print: error while loading shared libraries: libmosquitto.so.1: cannot open shared object file: No such file or directory

This is because the shared library file libmosquitto.so.1 cannot be found, causing the loading to fail.

In general, Linux searches for the required libraries in the /usr/lib path, and the libmosquitto.so.1 library is not in this path.

There are two solutions: one is to copy this file to the /usr/lib path, but this is generally not allowed, and ordinary users are also not permitted to copy things into this path. The other is to add the path where libmosquitto.so.1 is located to the dynamic library search path, the command is:

Swipe to view all code>>>

export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH

For explanations on this aspect, you can read previous notes: “Supplement on Static and Dynamic Linking (Linux)”.

After adding the dynamic library search path as mentioned above, both programs can run normally:

In the ParseResult.txt text file, we obtain:

Experiment successful!

This concludes this sharing, the code was written rather hastily, if there are any errors, please feel free to point them out, thank you! Since preparing the demo took a considerable amount of time and many comments were written.

Therefore, this article will not have too much explanation, interested friends can refer to this article’s demo and the two source files mosquitto/client/pub_client.c, mosquitto/client/sub_client.c.

The demo of this article:

Can be obtained by replying with the keyword: json_mqtt_demo in the chat interface of this public account. If you cannot obtain it, please contact me for acquisition.

Recommended Materials:https://blog.csdn.net/Dancer__Sky/article/details/77855249

Recently, the number of people joining the group has increased, both groups are full, now a third group has been created, interested parties can join by themselves: