کار با ماژول تمام عیار mc60 – قسمت هفتم – کار با MQTT

اگر که توی حوزه IOT کار میکنید یا حداقل یه بار از کنارش رد شده باشید به احتمال زیاد اسم MQTT به گوشتون خورده ، پروتکلی تحت شبکه که خوراک بحث اینترنت اشیا هست و این روزها همه دارند به سمتش میرند. توی این آموزش مختصری این پروتکل رو توضیح داده و سپس سراغ پیاده سازی یک پروژه به کمک اون میریم ، پس با ما همراه باشید.

MQTT چگونه کار میکند

نمیدونم چرا تا قبل از نوشتن این مطلب MQTT برام خیلی ترسناک بود و فکر میکردم خیلی سخته (البته سراغشم نرفته بودم) ! برای همین سعی میکنم خیلی ساده توضیح بدم تا برای همه قابل فهم باشه .

به تصویر بالا نگاه کنید نقطه شروع ما دماسنج سمت چپ هست ، این دماسنج یک کلاینت mqtt هست که به صورت publisher عمل میکنه ، یعنی دیتا ارسال میکنه ، به سروری که به اون broker میگیم. حالا دماسنج اطلاعات رو به آدرس خاصی در بروکر ارسال میکنه ، میتونیم اون آدرس رو مثل یک کانال تلگرامی درنظر بگیریم ، حالا کلاینت هایی که میخواند دیتا رو دریافت کنند ، توی این کانال (به آدرسی که در اون ، اطلاعات ارسال میشه) عضو میشند و هر موقع که یک دیتای جدید اومد متوجه میشند ، این دریافت دیتا رو کتابخونه ای که شما استفاده میکنید نحوه کارش رو مشخص میکنه ، مثلا به شما یک تابع میده که هرموقع دیتا اومد اون تابع فراخوانی میشه. هر کلاینت میتونه هم به عنوان publisher و هم subscriber (دریافت کننده اطلاعات) عمل کنه.

حالا بریم سراغ پروژه خودمون! ما میخوایم از یک ماژول mc60 به عنوان mqtt client استفاده کنیم و هم دیتا ارسال کنیم به سرور و هم دریافت کنیم (publisherوsubscriber). این کار رو هم میخوایم به صورت open cpu انجام بدیم نه با at command. بعد از اون هم داشبوردی که به کمک پلتفرم thingsboard طراحی میکنیم و دیتاهای دریافتی رو توی اون نمایش میدیم.

این هم از کد برنامه:

#define __CUSTOMER_CODE__
#ifdef __CUSTOMER_CODE__
#include "custom_feature_def.h"
#include "ql_stdlib.h"
#include "ql_common.h"
#include "ql_system.h"
#include "ql_type.h"
#include "ql_trace.h"
#include "ql_error.h"
#include "ql_uart.h"
#include "ql_timer.h"
#include "ril_network.h"
#include "ril_mqtt.h"
#include "ril.h"
#include "ril_util.h"
#include "ril_system.h"

#include "ql_iic.h"
#include "oled.h"
#include "sht20.h"

/// MQTT ////////////////////////////////////////////////////////////////////////
//define process state
typedef enum
{
STATE_NW_QUERY_STATE,
STATE_MQTT_CFG,
STATE_MQTT_OPEN,
STATE_MQTT_CONN,
STATE_MQTT_SUB,
STATE_MQTT_PUB,
STATE_MQTT_TUNS,
STATE_MQTT_CLOSE,
STATE_MQTT_DISC,
STATE_MQTT_TOTAL_NUM
} Enum_ONENETSTATE;
static u8 m_mqtt_state = STATE_NW_QUERY_STATE;
//timer
#define MQTT_TIMER_ID 0x200
#define MQTT_TIMER_PERIOD 500
//param
MQTT_Urc_Param_t *mqtt_urc_param_ptr = NULL;
ST_MQTT_topic_info_t mqtt_topic_info_t;
bool DISC_flag = TRUE;
bool CLOSE_flag = TRUE;

//connect info
Enum_ConnectID connect_id = ConnectID_0;
u8 clientID[] = "mc60\0";
u8 username[] = "XXXXXXXXXXXXXXX\0";
u8 passwd[] = "\0";
//topic and data
u32 pub_message_id = 0;
u32 sub_message_id = 0;
static u8 test_data[128] = "{\"temp\":5}\0"; //packet data
static u8 pu_topic[128] = "v1/devices/me/telemetry\0"; //Publisher topic
static u8 su_topic[128] = "v1/devices/me/attributes\0"; //Subscriber topic
////////////////////////////////////////////////////////////////////////////////////

///for GPRS ////////////////////////////////////////////////////////////////
//APN
#define APN "CMNET\0"
#define USERID ""
#define PASSWD ""
//SERVER
#define HOST_NAME "thingsboard.cloud"
#define HOST_PORT 1883
/////////////////////////////////////////////////////////////////////////////

/// DEBUG AND UART ///////////////////////////////////////////////////////////////////////////////////////////////////
#define SERIAL_RX_BUFFER_LEN 2048
#define DEBUG_PORT UART_PORT1
#define DBG_BUF_LEN 512
static char DBG_BUFFER[DBG_BUF_LEN];
#define APP_DEBUG(FORMAT, ...) \
{ \
Ql_memset(DBG_BUFFER, 0, DBG_BUF_LEN); \
Ql_sprintf(DBG_BUFFER, FORMAT, ##__VA_ARGS__); \
Ql_UART_Write((Enum_SerialPort)(DEBUG_PORT), (u8 *)(DBG_BUFFER), Ql_strlen((const char *)(DBG_BUFFER))); \
}

char m_RxBuf_Uart[SERIAL_RX_BUFFER_LEN];
static Enum_SerialPort m_myUartPort = UART_PORT1;
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

static void CallBack_UART_Hdlr(Enum_SerialPort port, Enum_UARTEventType msg, bool level, void *customizedPara);
static void Callback_Timer(u32 timerId, void *param);
static void mqtt_recv(u8 *buffer, u32 length);
static void message(u8 *data);

void proc_subtask1(s32 taskId)
{
u8 data[100] = "";
u8 temp[100] = "";
u32 step = 0;
u32 ret = 0;
while (1)
{
if (m_mqtt_state == STATE_MQTT_TOTAL_NUM)
{
step++;
Ql_sprintf(data, "{\"temp\":%5.2f}\0", readTemperature());
set1X();
setCursor(0, 3);
Ql_sprintf(temp, "Tem: %5.2f Hum: %5.2f \0", readTemperature(), readHumidity());
oledPrint(temp);
pub_message_id++; // The range is 0-65535. It will be 0 only when<qos>=0.
ret = RIL_MQTT_QMTPUB(connect_id, pub_message_id, QOS1_AT_LEASET_ONCE, 0, pu_topic, Ql_strlen(data), data);
if (RIL_AT_SUCCESS == ret)
{
APP_DEBUG("//<Start publish a message to server\r\n");
}
else
{
APP_DEBUG("//<Publish a message to server failure,ret = %d\r\n", ret);
}
}
Ql_Sleep(3000);
}
}

void proc_main_task(s32 taskId)
{
ST_MSG msg;
s32 ret;

//<Register & open UART port
Ql_UART_Register(m_myUartPort, CallBack_UART_Hdlr, NULL);
Ql_UART_Open(m_myUartPort, 115200, FC_NONE);

APP_DEBUG("START PROGRAM MQTT Client(SISOOG.COM)\r\n");

//<register state timer
Ql_Timer_Register(MQTT_TIMER_ID, Callback_Timer, NULL);

//register MQTT recv callback
ret = Ql_Mqtt_Recv_Register(mqtt_recv);
APP_DEBUG("//<register recv callback,ret = %d\r\n", ret);

//init i2c for oled and sht20
Ql_IIC_Init(0, PINNAME_RI, PINNAME_DCD, TRUE);
Ql_IIC_Config(0, TRUE, SLAVE_ADDRESS, 300);

//init oled display
oledInit();
oledClear();
setFont(Adafruit5x7);

while (TRUE)
{
Ql_OS_GetMessage(&msg);
switch (msg.message)
{
case MSG_ID_RIL_READY:
APP_DEBUG("//<RIL is ready\r\n");
Ql_RIL_Initialize();
break;
case MSG_ID_URC_INDICATION:
{
switch (msg.param1)
{
case URC_SIM_CARD_STATE_IND:
{
APP_DEBUG("//<SIM Card Status:%d\r\n", msg.param2);
if (SIM_STAT_READY == msg.param2)
{
Ql_Timer_Start(MQTT_TIMER_ID, MQTT_TIMER_PERIOD, TRUE);
APP_DEBUG("//<state timer start,ret = %d\r\n", ret);
}
}
break;
case URC_MQTT_OPEN:
{
mqtt_urc_param_ptr = msg.param2;
if (0 == mqtt_urc_param_ptr->result)
{
APP_DEBUG("//<Open a MQTT client successfully\r\n");
m_mqtt_state = STATE_MQTT_CONN;
}
else
{
APP_DEBUG("//<Open a MQTT client failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
case URC_MQTT_CONN:
{
mqtt_urc_param_ptr = msg.param2;
if (0 == mqtt_urc_param_ptr->result)
{
APP_DEBUG("//<Connect to MQTT server successfully\r\n");
m_mqtt_state = STATE_MQTT_SUB;
}
else
{
APP_DEBUG("//<Connect to MQTT server failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
case URC_MQTT_SUB:
{
mqtt_urc_param_ptr = msg.param2;
if ((0 == mqtt_urc_param_ptr->result) && (128 != mqtt_urc_param_ptr->sub_value[0]))
{
APP_DEBUG("//<Subscribe topics successfully\r\n");
m_mqtt_state = STATE_MQTT_PUB;
}
else
{
APP_DEBUG("//<Subscribe topics failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
case URC_MQTT_PUB:
{
mqtt_urc_param_ptr = msg.param2;
if (0 == mqtt_urc_param_ptr->result)
{
APP_DEBUG("//<Publish messages to MQTT server successfully\r\n");
m_mqtt_state = STATE_MQTT_TOTAL_NUM;
}
else
{
APP_DEBUG("//<Publish messages to MQTT server failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
case URC_MQTT_CLOSE:
{
mqtt_urc_param_ptr = msg.param2;
if (0 == mqtt_urc_param_ptr->result)
{
APP_DEBUG("//<Closed MQTT socket successfully\r\n");
}
else
{
APP_DEBUG("//<Closed MQTT socket failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
case URC_MQTT_DISC:
{
mqtt_urc_param_ptr = msg.param2;
if (0 == mqtt_urc_param_ptr->result)
{
APP_DEBUG("//<Disconnect MQTT successfully\r\n");
}
else
{
APP_DEBUG("//<Disconnect MQTT failure,error = %d\r\n", mqtt_urc_param_ptr->result);
}
}
break;
default:
break;
}
}
break;
default:
break;
}
}
}

static void CallBack_UART_Hdlr(Enum_SerialPort port, Enum_UARTEventType msg, bool level, void *customizedPara)
{
}

static void message(u8 *data)
{
set2X();
setCursor(10, 0);
Ql_sprintf(data, "%s \0", data);
oledPrint(data);
APP_DEBUG("message: %s\r\n", data);
}

static void mqtt_recv(u8 *buffer, u32 length)
{
APP_DEBUG("//<data:%s,len:%d\r\n", buffer, length);
u16 sec = 0;
for (u32 i = 0; i < 100; i++)
{
if (buffer[i] == ',')
{
sec++;
if (sec == 4)
{
u8 data[100] = "";
for (u32 y = 0; y < 100; y++)
{

data[y] = buffer[i + 10];
i++;
if (data[y] == '"')
{
data[y] = '\0';
message(data);
break;
break;
}
}
}
}
}
}

static void Callback_Timer(u32 timerId, void *param)
{
s32 ret;

if (MQTT_TIMER_ID == timerId)
{
switch (m_mqtt_state)
{
case STATE_NW_QUERY_STATE:
{
s32 cgreg = 0;
ret = RIL_NW_GetGPRSState(&cgreg);
set2X();
setCursor(10, 0);
oledPrint("Net check ");
APP_DEBUG("//<Network State:cgreg = %d\r\n", cgreg);
if ((cgreg == NW_STAT_REGISTERED) || (cgreg == NW_STAT_REGISTERED_ROAMING))
{
//<Set PDP context 0
RIL_NW_SetGPRSContext(0);
APP_DEBUG("//<Set PDP context 0 \r\n");
//<Set APN
ret = RIL_NW_SetAPN(1, APN, USERID, PASSWD);
APP_DEBUG("//<Set APN \r\n");
set2X();
setCursor(10, 0);
oledPrint("Set APN ");
//PDP activated
ret = RIL_NW_OpenPDPContext();
if (ret == RIL_AT_SUCCESS)
{
set2X();
setCursor(10, 0);
oledPrint("Act PDP ");
APP_DEBUG("//<Activate PDP context,ret = %d\r\n", ret);
m_mqtt_state = STATE_MQTT_CFG;
}
}
break;
}
case STATE_MQTT_CFG:
{
RIL_MQTT_QMTCFG_Showrecvlen(connect_id, ShowFlag_1); //<This sentence must be configured. The configuration will definitely succeed, so there is no need to care about.
ret = RIL_MQTT_QMTCFG_Version_Select(connect_id, Version_3_1_1);
if (RIL_AT_SUCCESS == ret)
{
APP_DEBUG("//<Select version 3.1.1 successfully\r\n");
m_mqtt_state = STATE_MQTT_OPEN;
}
else
{
APP_DEBUG("//<Select version 3.1.1 failure,ret = %d\r\n", ret);
}
break;
}
case STATE_MQTT_OPEN:
{
ret = RIL_MQTT_QMTOPEN(connect_id, HOST_NAME, HOST_PORT);
if (RIL_AT_SUCCESS == ret)
{
set2X();
setCursor(10, 0);
oledPrint("open MQTT ");
APP_DEBUG("//<Start opening a MQTT client\r\n");
if (FALSE == CLOSE_flag)
CLOSE_flag = TRUE;
m_mqtt_state = STATE_MQTT_TOTAL_NUM;
}
else
{
APP_DEBUG("//<Open a MQTT client failure,ret = %d-->\r\n", ret);
}
break;
}
case STATE_MQTT_CONN:
{
ret = RIL_MQTT_QMTCONN(connect_id, clientID, username, passwd);
if (RIL_AT_SUCCESS == ret)
{
set2X();
setCursor(10, 0);
oledPrint("con MQTT ");
APP_DEBUG("//<Start connect to MQTT server\r\n");
if (FALSE == DISC_flag)
DISC_flag = TRUE;
m_mqtt_state = STATE_MQTT_TOTAL_NUM;
}
else
{
APP_DEBUG("//<connect to MQTT server failure,ret = %d\r\n", ret);
}
break;
}
case STATE_MQTT_SUB:
{
mqtt_topic_info_t.count = 1;
mqtt_topic_info_t.topic[0] = (u8 *)Ql_MEM_Alloc(sizeof(u8) * 256);

Ql_memset(mqtt_topic_info_t.topic[0], 0, 256);
Ql_memcpy(mqtt_topic_info_t.topic[0], su_topic, Ql_strlen(su_topic));
mqtt_topic_info_t.qos[0] = QOS1_AT_LEASET_ONCE;
sub_message_id++; //< 1-65535.

ret = RIL_MQTT_QMTSUB(connect_id, sub_message_id, &mqtt_topic_info_t);

Ql_MEM_Free(mqtt_topic_info_t.topic[0]);
mqtt_topic_info_t.topic[0] = NULL;
if (RIL_AT_SUCCESS == ret)
{
set2X();
setCursor(10, 0);
oledPrint("sub MQTT ");
APP_DEBUG("//<Start subscribe topic\r\n");
m_mqtt_state = STATE_MQTT_TOTAL_NUM;
}
else
{
APP_DEBUG("//<Subscribe topic failure,ret = %d\r\n", ret);
}
break;
}
case STATE_MQTT_PUB:
{
pub_message_id++; //< The range is 0-65535. It will be 0 only when<qos>=0.
ret = RIL_MQTT_QMTPUB(connect_id, pub_message_id, QOS1_AT_LEASET_ONCE, 0, pu_topic, Ql_strlen(test_data), test_data);
if (RIL_AT_SUCCESS == ret)
{
APP_DEBUG("//<Start publish a message to MQTT server\r\n");
m_mqtt_state = STATE_MQTT_TOTAL_NUM;
}
else
{
APP_DEBUG("//<Publish a message to MQTT server failure,ret = %d\r\n", ret);
}
break;
}
case STATE_MQTT_TOTAL_NUM:
{
break;
}
default:
break;
}
}
}
#endif // __CUSTOMER_CODE__

همچنین لازمه که یک تسک به نام proc_subtask1 اضافه بشه به برنامه و کتابخانه های sht20 و oled هم به پروژه اضافه بشه.

توضیح کلیات کد

توی کد ما دو تسک اجرا میشه تسک اصلی وظیه راه اندازی بخش های مختلف و oled و … رو داره ، تسک دوم هم در صورتی که به سرور متصل باشیم هر 3 ثانیه دما رو برای سرور ارسال میکنه.

غیر از اون هم یه تایمر داریم که هر 500 میلی ثانیه اجرا میشه و مرحله به مرحله اتصال به سرور و راه اندازی mqtt رو انجام میده.

بعد از اتصال به سرور و subscribe شدن ماژول ، بعد از هر بار دریافت دیتا در تاپیک مورد نظر تابع mqtt_recv صدا زده میشه ، من توی این تابع کدی نوشتم که بیاد و متن دریافت شده رو برای تابع message ارسال کنه تا هم روی نمایشگر نشون داده بشه و هم توی سریال چاپ بشه.

منبع:سیسوگ