OpenHarmony開發(fā)之MQTT講解

//訂閱的回調(diào)函數(shù)
void messageArrived(MessageData *data)
{
    printf("Message arrived on topic %.*s: %.*s
", data->topicName->lenstring.len,
        data->topicName->lenstring.data, data->message->payloadlen, data->message->payload);
}
//主流程函數(shù)
static void MQTTDemoTask(void)
{
    WifiConnect("BearPi", "123456789");
    printf("Starting ...
");
    int rc, count = 0;
    MQTTClient client;


    NetworkInit(&network);
    printf("NetworkConnect  ...
");


    NetworkConnect(&network, MQTT_SERVERIP, MQTT_SERVERPORT);//本地電腦作為消息代理 此處為電腦IP
    printf("MQTTClientInit  ...
");
    //1-------------創(chuàng)建客戶端對象
    MQTTClientInit(&client, &network, MQTT_CMD_TIMEOUT_MS, sendBuf, sizeof(sendBuf), readBuf, sizeof(readBuf));


    MQTTString clientId = MQTTString_initializer;
    clientId.cstring = "bearpi";


    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    data.clientID = clientId;
    data.willFlag = 0;
    data.MQTTVersion = MQTT_VERSION;
    data.keepAliveInterval = MQTT_KEEP_ALIVE_MS;
    data.cleansession = 1;


    printf("MQTTConnect  ...
");
    //2-------------連接服務端
    rc = MQTTConnect(&client, &data);
    if (rc != 0) {
        printf("MQTTConnect: %d
", rc);
        NetworkDisconnect(&network);
        MQTTDisconnect(&client);
        osDelay(MQTT_DELAY_2S);
    }


    printf("MQTTSubscribe  ...
");
    //3-------------訂閱主題substopic
    rc = MQTTSubscribe(&client, "substopic", MQTT_QOS, messageArrived);
    if (rc != 0) {
        printf("MQTTSubscribe: %d
", rc);
        osDelay(MQTT_DELAY_2S);
    }
    while (++count) {
        MQTTMessage message;
        char payload[30];


        message.qos = MQTT_QOS;
        message.retained = 0;
        message.payload = payload;
        (void)sprintf_s(payload, sizeof(payload), "message number %d", count);
        message.payloadlen = strlen(payload);
    //4------------發(fā)布pubtopic主題
        if ((rc = MQTTPublish(&client, "pubtopic", &message)) != 0) {
            printf("Return code from MQTT publish is %d
", rc);
            NetworkDisconnect(&network);
            MQTTDisconnect(&client);
        }
        osDelay(MQTT_DELAY_500_MS);
    }
}

NetworkConnect(&network, 192.168.120.137, 1883);//本地電腦作為消息代理 此處為電腦IP Mosquitto相匹配

printf("Starting ...
");
NetworkInit(&network);
printf("NetworkConnect  ...
");
NetworkConnect(&network, MQTT_SERVERIP, MQTT_SERVERPORT);//本地電腦作
printf("MQTTClientInit  ...
");
//1-------------創(chuàng)建客戶端對象
MQTTClientInit(&client, &network, MQTT_CMD_TIMEOUT_MS, sendBuf, sizeof(sendBuf), readBuf, sizeof(readBuf));


printf("MQTTConnect  ...
");
//2-------------連接服務端
rc = MQTTConnect(&client, &data);
    
printf("MQTTSubscribe  ...
");
//3-------------訂閱主題substopic
rc = MQTTSubscribe(&client, "substopic", MQTT_QOS, messageArrived);
(void)sprintf_s(payload, sizeof(payload), "message number %d", count);
 //4------------循環(huán)發(fā)布pubtopic主題 內(nèi)容為message number+connt的計數(shù)值
 MQTTPublish(&client, "pubtopic", &message)
//訂閱的回調(diào)函數(shù)輸出以下內(nèi)容
printf("Message arrived on topic %.*s: %.*s
", data->topicName->lenstring.len,
data->topicName->lenstring.data,data->message->payloadlen,data->message->payload);