1 阿里云物联网平台

在设备管理下创建两个设备，参考ESP32基础应用之ESP32与阿里云物联网平台实现数据互传（MQTT协议）
在规则引擎下创建云产品流转，包括数据源、数据目的、解析器，参考云产品流转概述
创建数据源

创建数据目的
创建解析器
依次关联数据源、数据目的、解析脚本，解析脚本较为复杂，这里只实现简单的数据转发到另一Topic
```
//通过payload函数，获取设备上报的消息内容，并按照JSON格式转换。
var data = payload("json");
//直接流转物模型上报数据。
writeIotTopic(1000, "/a1Buwr73pUI/dev-esp32-com5/user/get", data)
```
这里只实现了设备1将数据流转到设备2，还需要使用同样的方法创建另一个解析器将设备2的数据流转到设备1。
云产品流转有许多复杂的规则与应用，这里只实现简单的数据转发到另一Topic

2 程序编写

程序逻辑：

ESP32-COM4 设备按键按下，上传数据到阿里云物联网平台并流传到ESP32-COM5设备，将数据打印出来；
ESP32-COM5 设备按键按下，上传数据到阿里云物联网平台并流传到ESP32-COM4设备，将数据打印出来；

2.1 ESP32-COM4 设备

本地数据上传

static void mqtt_test_task(void *pvParameters)
{
    int  msg_id;
    char key_flag = 0;
    //构造JSON格式数据，该数据用于反馈给阿里云物联网平台，
    cJSON *Wroot =	cJSON_CreateObject();
    cJSON_AddItemToObject(Wroot, "DeviceSource", cJSON_CreateString("ESP32-COM4"));
    cJSON_AddItemToObject(Wroot, "DevicePurpose", cJSON_CreateString("ESP32-COM5"));
    cJSON_AddItemToObject(Wroot, "Data", cJSON_CreateString("hello ,I am COM4"));
    cJSON_AddItemToObject(Wroot, "Date", cJSON_CreateString("444444"));
    memset(local_data_tx,0x0,LOCAL_DATA_SIZE);		
    int len = strlen(cJSON_Print(Wroot));			
    memcpy(local_data_tx, cJSON_Print(Wroot),len); //将JSON格式数据复制到数组中，将以数组的形式传递给发布函数
    while(1)
    {
        if(0 == gpio_get_level(USER_KEY_PIN)){
            vTaskDelay(10 / portTICK_RATE_MS);
            while(0 == gpio_get_level(USER_KEY_PIN)){
                if(key_flag == 0){
                    key_flag = 1;
                    printf("key press  \n");
                    msg_id = esp_mqtt_client_publish(client, AliyunPublishTopic_user_update, local_data_tx, strlen(local_data_tx), 0, 0);
                    ESP_LOGI(TAG, "esp32 com4 send data successful, msg_id=%d", msg_id);	
                }
            }
	    }else{
            key_flag = 0;
        }
    }
    vTaskDelete(NULL);
}

打印收到的数据

static int user_parse_json(char *json_data)
{
	cJSON *root = NULL;
    root = cJSON_Parse(json_data);   
    printf("%s\n\n", cJSON_Print(root));   /*将完整的数据以JSON格式打印出来*/
    return 0;
}
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event)
{
    esp_mqtt_client_handle_t client = event->client;
    int  msg_id;
    switch (event->event_id) {
        case MQTT_EVENT_DATA:
            ESP_LOGI(TAG, "MQTT_EVENT_DATA");
            printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
            printf("DATA=%.*s\r\n", event->data_len, event->data);
            memset(local_data_rx,0x0,LOCAL_DATA_SIZE);
            strncpy(local_data_rx, event->data, event->data_len); //将指针类型的数据复制到一个数组中
            msg_id = user_parse_json(local_data_rx);
            break;
        default:
            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
            break;
    }
    return ESP_OK;
}

2.1 ESP32-COM5 设备

本地数据上传

static void mqtt_test_task(void *pvParameters)
{
    int  msg_id;
    char key_flag = 0;
    //构造JSON格式数据，该数据用于反馈给阿里云物联网平台，
    cJSON *Wroot =	cJSON_CreateObject();
    cJSON_AddItemToObject(Wroot, "DeviceSource", cJSON_CreateString("ESP32-COM5"));
    cJSON_AddItemToObject(Wroot, "DevicePurpose", cJSON_CreateString("ESP32-COM4"));
    cJSON_AddItemToObject(Wroot, "Data", cJSON_CreateString("hello ,I am COM5"));
    cJSON_AddItemToObject(Wroot, "Date", cJSON_CreateString("555555"));
    memset(local_data_tx,0x0,LOCAL_DATA_SIZE);		
    int len = strlen(cJSON_Print(Wroot));			
    memcpy(local_data_tx, cJSON_Print(Wroot),len); //将JSON格式数据复制到数组中，将以数组的形式传递给发布函数

    for(;;)
    {
        if(0 == gpio_get_level(USER_KEY_PIN)){
            vTaskDelay(10 / portTICK_RATE_MS);
            while(0 == gpio_get_level(USER_KEY_PIN)){
                if(key_flag == 0){
                    key_flag = 1;
                    printf("key press  \n");
                    msg_id = esp_mqtt_client_publish(client, AliyunPublishTopic_user_update, local_data_tx, strlen(local_data_tx), 0, 0);
                    ESP_LOGI(TAG, "esp32 com4 send data successful, msg_id=%d", msg_id);	
                }
            }
	    }else{
            key_flag = 0;
        }
	}
    vTaskDelete(NULL);
}

打印收到的数据

static int user_parse_json(char *json_data)
{
	cJSON *root = NULL;
    root = cJSON_Parse(json_data);   
    printf("%s\n\n", cJSON_Print(root));   /*将完整的数据以JSON格式打印出来*/
    return 0;
}
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event)
{
    esp_mqtt_client_handle_t client = event->client;
    int  msg_id;
    switch (event->event_id) {
        case MQTT_EVENT_DATA:
            ESP_LOGI(TAG, "MQTT_EVENT_DATA");
            printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
            printf("DATA=%.*s\r\n", event->data_len, event->data);
            memset(local_data_rx,0x0,LOCAL_DATA_SIZE);
            strncpy(local_data_rx, event->data, event->data_len); //将指针类型的数据复制到一个数组中
            msg_id = user_parse_json(local_data_rx);
            break;
        default:
            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
            break;
    }
    return ESP_OK;
}