#include "common.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #define MAX_VCOM 4 #define ETH_0_NAME "eth0" #define ETH_1_NAME "eth1" #define GET_LAN_IP "ifconfig eth0 | grep 'inet addr' | awk -F'[ :]+' '{print $4}'" #define GET_LAN_NETMASK "ifconfig eth0 | grep 'inet addr' | awk -F'[ :]+' '{print $8}'" #define GET_LAN_GATEWAY "ifconfig eth0 | grep 'inet addr' | awk -F'[ :]+' '{print $6}'" #define GET_LAN_MACADDR "ifconfig eth0 | grep 'HWaddr' | awk -F' ' '{print $5}'" #define GET_LAN_STATE "cat /sys/class/net/eth0/operstate" #define GET_WAN_IP "ifconfig eth1 | grep 'inet addr' | awk -F'[ :]+' '{print $4}'" #define GET_WAN_NETMASK "ifconfig eth1 | grep 'inet addr' | awk -F'[ :]+' '{print $8}'" #define GET_WAN_GATEWAY "ifconfig eth1 | grep 'inet addr' | awk -F'[ :]+' '{print $6}'" #define GET_WAN_MACADDR "ifconfig eth1 | grep 'HWaddr' | awk -F' ' '{print $5}'" #define GET_WAN_STATE "cat /sys/class/net/eth1/operstate" #define GET_LTE_IP "ifconfig usb0 | grep 'inet addr' | awk -F'[ :]+' '{print $4}'" #define GET_LTE_NETMASK "ifconfig usb0 | grep 'inet addr' | awk -F'[ :]+' '{print $8}'" #define GET_LTE_GATEWAY "ifconfig usb0 | grep 'inet addr' | awk -F'[ :]+' '{print $6}'" #define GET_LTE_MACADDR "ifconfig usb0 | grep 'HWaddr' | awk -F' ' '{print $5}'" #define GET_LTE_STATE "cat /sys/class/net/usb0/operstate" #define GET_WLAN_IP "ifconfig wlan0 | grep 'inet addr' | awk -F'[ :]+' '{print $4}'" #define GET_WLAN_NETMASK "ifconfig wlan0 | grep 'inet addr' | awk -F'[ :]+' '{print $8}'" #define GET_WLAN_GATEWAY "ifconfig wlan0 | grep 'inet addr' | awk -F'[ :]+' '{print $6}'" #define GET_WLAN_MACADDR "ifconfig wlan0 | grep 'HWaddr' | awk -F' ' '{print $5}'" #define GET_WLAN_STATE "cat /sys/class/net/wlan0/operstate" #define GET_TOTAL_MEM "free -m | grep 'Mem:' | awk -F' ' '{print $2}'" #define GET_USED_MEM "free -m | grep 'Mem:' | awk -F' ' '{print $3}'" #define GET_TOTAL_FLASH "4096" #define GET_USED_FLASH "362" #define GET_NOWTIME "date +'%Y-%m-%d %H:%M.%S'" int vcom_fd[MAX_VCOM]; struct mosquitto* mosq_up_mqtt[MAX_VCOM]; struct mosquitto* mosq_dw_mqtt[MAX_VCOM]; volatile bool exit_sig = false; volatile bool quit_sig = false; int net_sock = 0; struct json_object* root_obj = NULL; pthread_mutex_t mx_db = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t root_lock; struct mosquitto* mosq_http; static uint16_t CRC16(const uint8_t* p_data, uint32_t len, uint16_t* p_crc) { uint32_t i; uint16_t crc = 0xffff; if (NULL == p_data || NULL == p_crc || 0 == len) { return -1; } for (i = 0; i < len; i++) { crc = (unsigned char)(crc >> 8) | (crc << 8); crc ^= p_data[i]; crc ^= (unsigned char)(crc & 0xff) >> 4; crc ^= (crc << 8) << 4; crc ^= ((crc & 0xff) << 4) << 1; } *p_crc = crc; return 0; } void exec_cmd(char* cmd, char* buf) { FILE* fp = NULL; if ((fp = popen(cmd, "r")) == NULL) { printf("error: popen failed!\r\n"); exit(1); } while (fgets(buf, 199, fp) != NULL) { if (buf[strlen(buf) - 1] == '\n') buf[strlen(buf) - 1] = '\0'; } pclose(fp); } static void sig_handler(int sigio) { if (sigio == SIGQUIT) { quit_sig = true; } else if ((sigio == SIGINT) || (sigio == SIGTERM)) { exit_sig = true; } return; } /* modbus ---> json { "reg_0001":23, "reg_0002":24 } */ int dwlink_packet_process(uint8_t* device_id_str, uint8_t* dw_buffer, uint8_t* dw_buffer_len, uint8_t* attr_addr_str, uint8_t* attr_value_str) { struct json_object* dw_obj = NULL; dw_obj = json_object_new_object(); if (dw_obj == NULL) { printf("dw_obj is null\r\n"); return NULL; } json_object_object_add(dw_obj, attr_addr_str, json_object_new_string(attr_value_str)); if (strlen(json_object_to_json_string(dw_obj)) < 255) { memcpy(dw_buffer, json_object_to_json_string(dw_obj), strlen(json_object_to_json_string(dw_obj))); *dw_buffer_len = strlen(json_object_to_json_string(dw_obj)); } json_object_put(dw_obj); } void config_init() { root_obj = json_object_from_file("./app.json"); if (root_obj == NULL) { return 0; } } void mosq_http_connect_callback(struct mosquitto* mosq, void* obj, int rc) { printf("http_json_connect_callback set\r\n"); } void mosq_http_disconnect_callback(struct mosquitto* mosq, void* obj, int result) { printf("http_json_disconnect_callback set\r\n"); } struct json_object* get_system_json() { struct json_object *rsp_obj = NULL, *system_obj = NULL; rsp_obj = json_object_new_object(); if (rsp_obj == NULL) { printf("rsp_obj is null\r\n"); return NULL; } json_object_object_get_ex(root_obj, "system", &system_obj); json_object_object_add(rsp_obj, "system", system_obj); json_object_get(system_obj); json_object_object_add(rsp_obj, "state", json_object_new_string("success")); return rsp_obj; } struct json_object* get_platform_json() { struct json_object *rsp_obj = NULL, *platform_obj = NULL; rsp_obj = json_object_new_object(); if (rsp_obj == NULL) { printf("rsp_obj is null\r\n"); return NULL; } json_object_object_get_ex(root_obj, "platform", &platform_obj); json_object_object_add(rsp_obj, "platform", platform_obj); json_object_get(platform_obj); json_object_object_add(rsp_obj, "state", json_object_new_string("success")); return rsp_obj; } struct json_object* get_user_json() { struct json_object *rsp_obj = NULL, *user_obj = NULL; rsp_obj = json_object_new_object(); if (rsp_obj == NULL) { printf("rsp_obj is null\r\n"); return NULL; } json_object_object_get_ex(root_obj, "user", &user_obj); json_object_object_add(rsp_obj, "user", user_obj); json_object_get(user_obj); json_object_object_add(rsp_obj, "state", json_object_new_string("success")); return rsp_obj; } struct json_object* get_devices_json() { struct json_object* rsp_obj = NULL; select_all_device(&rsp_obj); return rsp_obj; } struct json_object* set_devices_json(char* payload) { struct json_object *payload_obj = NULL, *item_obj = NULL, *devices_obj = NULL, *device_id_obj = NULL, *device_name_obj = NULL, *device_addr_obj = NULL, *application_id_obj = NULL; char *device_id_str = NULL, *device_name_str = NULL, *device_addr_str = NULL, *application_id_str = NULL; int len = 0, index = 0; payload_obj = json_tokener_parse(payload); if (payload_obj == NULL) { printf("payload is null\r\n"); return NULL; } json_object_object_get_ex(payload_obj, "devices", &devices_obj); len = json_object_array_length(devices_obj); for (index = 0; index < len; index++) { item_obj = json_object_array_get_idx(devices_obj, index); json_object_object_get_ex(item_obj, "device_id", &device_id_obj); device_id_str = json_object_get_string(device_id_obj); json_object_object_get_ex(item_obj, "device_name", &device_name_obj); device_name_str = json_object_get_string(device_name_obj); json_object_object_get_ex(item_obj, "device_addr", &device_addr_obj); device_addr_str = json_object_get_string(device_addr_obj); json_object_object_get_ex(item_obj, "application_id", &application_id_obj); application_id_str = json_object_get_string(application_id_obj); pthread_mutex_lock(&mx_db); update_device_by_device_id(tolower(device_id_str), tolower(device_name_str), atoi(device_addr_str), tolower(application_id_str)); pthread_mutex_unlock(&mx_db); } json_object_object_add(payload_obj, "state", json_object_new_string("success")); return payload_obj; } struct json_object* set_platform_json(char* payload) { struct json_object *payload_obj = NULL, *platform_obj = NULL; payload_obj = json_tokener_parse(payload); if (payload_obj == NULL) { printf("payload is null\r\n"); return NULL; } json_object_object_get_ex(payload_obj, "platform", &platform_obj); json_object_object_add(root_obj, "platform", platform_obj); json_object_get(platform_obj); json_object_to_file_ext("./app.json", root_obj, JSON_C_TO_STRING_PRETTY); sync(); json_object_object_add(payload_obj, "state", json_object_new_string("success")); return payload_obj; } struct json_object* del_devices_json(char* payload) { struct json_object *payload_obj = NULL, *item_obj = NULL, *devices_obj = NULL, *device_id_obj = NULL, *device_name_obj = NULL, *device_addr_obj = NULL, *module_id_obj = NULL; char *device_id_str = NULL, *device_name_str = NULL, *device_addr_str = NULL, *module_id_str = NULL; int len = 0, index = 0; payload_obj = json_tokener_parse(payload); if (payload_obj == NULL) { printf("payload is null\r\n"); return NULL; } json_object_object_get_ex(payload_obj, "devices", &devices_obj); len = json_object_array_length(devices_obj); for (index = 0; index < len; index++) { item_obj = json_object_array_get_idx(devices_obj, index); json_object_object_get_ex(item_obj, "device_id", &device_id_obj); device_id_str = json_object_get_string(device_id_obj); pthread_mutex_lock(&mx_db); delete_device_by_device_id(device_id_str); pthread_mutex_unlock(&mx_db); } json_object_object_add(payload_obj, "state", json_object_new_string("success")); return payload_obj; } struct json_object* add_devices_json(char* payload) { struct json_object *payload_obj = NULL, *item_obj = NULL, *devices_obj = NULL, *device_id_obj = NULL, *device_name_obj = NULL, *device_addr_obj = NULL, *application_id_obj = NULL; char *device_id_str = NULL, *device_name_str = NULL, *device_addr_str = NULL, *application_id_str = NULL; int len = 0, index = 0; int ret = 0; payload_obj = json_tokener_parse(payload); if (payload_obj == NULL) { printf("payload is null\r\n"); return NULL; } json_object_object_get_ex(payload_obj, "devices", &devices_obj); len = json_object_array_length(devices_obj); for (index = 0; index < len; index++) { item_obj = json_object_array_get_idx(devices_obj, index); json_object_object_get_ex(item_obj, "device_id", &device_id_obj); device_id_str = json_object_get_string(device_id_obj); json_object_object_get_ex(item_obj, "device_name", &device_name_obj); device_name_str = json_object_get_string(device_name_obj); json_object_object_get_ex(item_obj, "device_addr", &device_addr_obj); device_addr_str = json_object_get_string(device_addr_obj); json_object_object_get_ex(item_obj, "application_id", &application_id_obj); application_id_str = json_object_get_string(application_id_obj); pthread_mutex_lock(&mx_db); ret = insert_device_item(tolower(device_id_str), tolower(device_name_str), atoi(device_addr_str), tolower(application_id_str)); pthread_mutex_unlock(&mx_db); } if (ret == 0) json_object_object_add(payload_obj, "state", json_object_new_string("success")); else json_object_object_add(payload_obj, "state", json_object_new_string("fail")); return payload_obj; } // req:get/devices/none/http // rsp:ack/get/device/none/http void mosq_http_json_message_callback(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg) { struct json_object* rsp_obj = NULL; char** topics; int topic_count; char pub_topic[200]; mosquitto_sub_topic_tokenise(msg->topic, &topics, &topic_count); if (strncmp(topics[0], "get", STR_MAX_LEN(topics[0], "get")) == 0) { if (strncmp(topics[1], "system", STR_MAX_LEN(topics[1], "system")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = get_system_json(); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("get system rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } else if (strncmp(topics[1], "platform", STR_MAX_LEN(topics[1], "platform")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = get_platform_json(); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("get platform rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } else if (strncmp(topics[1], "user", STR_MAX_LEN(topics[1], "user")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = get_user_json(); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("get user rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } else if (strncmp(topics[1], "devices", STR_MAX_LEN(topics[1], "devices")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = get_devices_json(); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("get devices rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } } else if (strncmp(topics[0], "set", STR_MAX_LEN(topics[0], "set")) == 0) { if (strncmp(topics[1], "platform", STR_MAX_LEN(topics[1], "platform")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = set_platform_json(msg->payload); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("set platform rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } else if (strncmp(topics[1], "devices", STR_MAX_LEN(topics[1], "devices")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = set_devices_json(msg->payload); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("set devices rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } } else if (strncmp(topics[0], "del", STR_MAX_LEN(topics[0], "del")) == 0) { if (strncmp(topics[1], "devices", STR_MAX_LEN(topics[1], "devices")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = del_devices_json(msg->payload); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("del devices rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } } else if (strncmp(topics[0], "add", STR_MAX_LEN(topics[0], "add")) == 0) { if (strncmp(topics[1], "devices", STR_MAX_LEN(topics[1], "devices")) == 0) { pthread_mutex_lock(&root_lock); rsp_obj = add_devices_json(msg->payload); pthread_mutex_unlock(&root_lock); if (rsp_obj != NULL) { printf("add devices rsp_obj:%s\r\n", json_object_to_json_string(rsp_obj)); } } } if (rsp_obj != NULL) { sprintf(pub_topic, "ack/%s/%s/%s/%s", topics[0], topics[1], topics[2], topics[3]); mosquitto_publish(mosq, NULL, pub_topic, strlen(json_object_to_json_string(rsp_obj)), json_object_to_json_string(rsp_obj), 1, false); json_object_put(rsp_obj); } mosquitto_sub_topic_tokens_free(&topics, topic_count); printf("mosq_http_json_message_callback\r\n"); } void mosq_http_json_subscribe_callback(struct mosquitto* mosq, void* obj, int mid, int qos_count, const int* granted_qos) { printf("http_json_subscribe_callback set\r\n"); } // req:get/devices/none/http // rsp:ack/get/device/none/http void http_json_process_thread(void) { char pub_topic[200]; char id[50]; snprintf(id, 50, "http_json_%d", getpid()); mosq_http = mosquitto_new(id, true, NULL); mosquitto_connect_callback_set(mosq_http, mosq_http_connect_callback); mosquitto_disconnect_callback_set(mosq_http, mosq_http_disconnect_callback); mosquitto_message_callback_set(mosq_http, mosq_http_json_message_callback); mosquitto_subscribe_callback_set(mosq_http, mosq_http_json_subscribe_callback); mosquitto_connect(mosq_http, "localhost", 1883, 600); // mosquitto_username_pw_set(mosq_http,"leo_mqtt", "leo_mqtt@"); mosquitto_subscribe(mosq_http, NULL, "+/+/+/http", 1); while (!exit_sig && !quit_sig) { if (mosquitto_loop(mosq_http, 100, 1)) { mosquitto_reconnect(mosq_http); sleep(10); } usleep(500000); } printf("http_json_process_thread exit\r\n"); } int get_system_status() { char buf[200]; struct json_object* system_obj = NULL; json_object_object_get_ex(root_obj, "system", &system_obj); memset(buf, '\0', 200); exec_cmd(GET_WAN_MACADDR, buf); // printf("WAN-ADDR:%s\r\n",buf); json_object_object_add(system_obj, "wan_mac", json_object_new_string(buf)); memset(buf, '\0', 200); exec_cmd(GET_WAN_STATE, buf); // printf("WAN-STTE:%s\r\n",buf); if (strncmp(buf, "up", STR_MAX_LEN("up", buf)) == 0) { json_object_object_add(system_obj, "ethernet_wan", json_object_new_string("连接")); } else { json_object_object_add(system_obj, "ethernet_wan", json_object_new_string("断开")); } memset(buf, '\0', 200); exec_cmd(GET_LAN_MACADDR, buf); // printf("LAN-ADDR:%s\r\n",buf); json_object_object_add(system_obj, "lan_mac", json_object_new_string(buf)); memset(buf, '\0', 200); exec_cmd(GET_LAN_STATE, buf); // printf("LAN-STATE:%s\r\n",buf); if (strncmp(buf, "up", STR_MAX_LEN("up", buf)) == 0) { json_object_object_add(system_obj, "ethernet_lan", json_object_new_string("连接")); } else { json_object_object_add(system_obj, "ethernet_lan", json_object_new_string("断开")); } memset(buf, '\0', 200); exec_cmd(GET_LTE_MACADDR, buf); // printf("LTE-ADDR:%s\r\n",buf); json_object_object_add(system_obj, "lte_mac", json_object_new_string(buf)); memset(buf, '\0', 200); exec_cmd(GET_LTE_STATE, buf); // printf("LTE-STATE:%s\r\n",buf); if (strncmp(buf, "up", STR_MAX_LEN("up", buf)) == 0) { json_object_object_add(system_obj, "lte_status", json_object_new_string("连接")); } else { json_object_object_add(system_obj, "lte_status", json_object_new_string("断开")); } memset(buf, '\0', 200); exec_cmd(GET_TOTAL_MEM, buf); // printf("TOTAL-MEM:%s\r\n",buf); json_object_object_add(system_obj, "total_memory", json_object_new_string(buf)); memset(buf, '\0', 200); exec_cmd(GET_USED_MEM, buf); // printf("USED-MEM:%s\r\n",buf); json_object_object_add(system_obj, "used_memory", json_object_new_string(buf)); memset(buf, '\0', 200); // exec_cmd(GET_TOTAL_FLASH, buf); json_object_object_add(system_obj, "total_flash", json_object_new_string("4096")); memset(buf, '\0', 200); // exec_cmd(GET_USED_FLASH, buf); json_object_object_add(system_obj, "used_flash", json_object_new_string("361")); memset(buf, '\0', 200); exec_cmd(GET_NOWTIME, buf); json_object_object_add(system_obj, "boot_time", json_object_new_string(buf)); } void mosq_up_mqtt_connect_callback(struct mosquitto* mosq, void* obj, int rc) { printf("mosq_up_mqtt_connect_callback set\r\n"); } void mosq_up_mqtt_disconnect_callback(struct mosquitto* mosq, void* obj, int result) { printf("mosq_up_mqtt_disconnect_callback set\r\n"); } void mosq_up_mqtt_message_callback(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg) { struct json_object *payload_obj = NULL, *devEUI_obj = NULL, *object_obj = NULL; char* devEUI_str = NULL; payload_obj = json_tokener_parse(msg->payload); if (payload_obj == NULL) goto _exit_0; json_object_object_get_ex(payload_obj, "devEUI", &devEUI_obj); devEUI_str = json_object_get_string(devEUI_obj); json_object_object_get_ex(payload_obj, "object", &object_obj); if (object_obj == NULL || object_obj == "") { goto _exit_0; } json_object_object_foreach(object_obj, key, val) { printf("key:%s --- value:%s \r\n", key, json_object_to_json_string(val)); update_attrtab_item_by_device_id(devEUI_str, key, json_object_to_json_string(val)); } json_object_put(payload_obj); _exit_0: printf("mqtt_process_message_callback\r\n"); } void mosq_up_mqtt_subscribe_callback(struct mosquitto* mosq, void* obj, int mid, int qos_count, const int* granted_qos) { printf("mosq_up_mqtt_subscribe_callback set\r\n"); } /* uplink topic: application/+/device/+/event/up {"applicationID":"1","applicationName":"leo","deviceName":"1000000000000001","deviceProfileName":"leo_otaa_c","deviceProfileID":"172bf053-caf5-49d1-8bdb-5d989aef9ec1","devEUI":"1000000000000001","rxInfo":[{"gatewayID":"aa555a0000000000","uplinkID":"312238b9-ffc3-4a4d-914d-e8ace8ffbae1","name":"AA555A0000000000","rssi":-93,"loRaSNR":8.2,"location":{"latitude":0,"longitude":0,"altitude":0}}],"txInfo":{"frequency":476100000,"dr":2},"adr":false,"fCnt":3,"fPort":2,"data":"MTIzNDU2","object":{"reg_0001":49,"reg_0002":50}} dwlink topic: application/+/device/+/command/down */ void mqtt_modbus_uplink_process_thread(void* argc) { char sub_topic[200]; char id[50]; int mqtt_index = *((int*)argc); printf("mqtt_index_uplink:%d \r\n", mqtt_index); snprintf(id, 50, "mqtt_modbus_up_%d", mqtt_index); printf("mqtt_uplink_id:%s\r\n", id); mosq_up_mqtt[mqtt_index] = mosquitto_new(id, true, NULL); mosquitto_connect_callback_set(mosq_up_mqtt[mqtt_index], mosq_up_mqtt_connect_callback); mosquitto_disconnect_callback_set(mosq_up_mqtt[mqtt_index], mosq_up_mqtt_disconnect_callback); mosquitto_message_callback_set(mosq_up_mqtt[mqtt_index], mosq_up_mqtt_message_callback); mosquitto_subscribe_callback_set(mosq_up_mqtt[mqtt_index], mosq_up_mqtt_subscribe_callback); mosquitto_connect(mosq_up_mqtt[mqtt_index], "localhost", 1883, 600); sprintf(sub_topic, "application/%d/device/+/event/up", mqtt_index + 1); mosquitto_username_pw_set(mosq_up_mqtt[mqtt_index], "leo_mqtt", "leo_mqtt@"); mosquitto_subscribe(mosq_up_mqtt[mqtt_index], NULL, sub_topic, 0); while (!exit_sig && !quit_sig) { if (mosquitto_loop(mosq_up_mqtt[mqtt_index], 100, 1)) { mosquitto_reconnect(mosq_up_mqtt[mqtt_index]); sleep(10); } usleep(500000); } mosquitto_disconnect(mosq_up_mqtt[mqtt_index]); mosquitto_destroy(mosq_up_mqtt[mqtt_index]); printf("mqtt_modbus_uplink_process_thread exit\r\n"); } void mosq_dw_mqtt_connect_callback(struct mosquitto* mosq, void* obj, int rc) { printf("mosq_dw_mqtt_connect_callback set\r\n"); } void mosq_dw_mqtt_disconnect_callback(struct mosquitto* mosq, void* obj, int result) { printf("mosq_dw_mqtt_disconnect_callback set\r\n"); } void mosq_dw_mqtt_message_callback(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg) { printf("mosq_dw_mqtt_message_callback\r\n"); } void mosq_dw_mqtt_subscribe_callback(struct mosquitto* mosq, void* obj, int mid, int qos_count, const int* granted_qos) { printf("mosq_dw_mqtt_subscribe_callback set\r\n"); } /* dwlink topic: application/+/device/+/command/down */ void mqtt_modbus_dwlink_process_thread(void* argc) { uint8_t rx_buffer[255] = {0}; uint8_t tx_buffer[255] = {0}; int ret = 0, i = 0, j = 0, rx_buffer_len = 0, tx_buffer_len = 0; char attr_addr_str[9], attr_value_str[5]; uint16_t attr_value = 0, attr_num = 0, attr_addr = 0, crc = 0; uint8_t device_addr = 0, device_funcode = 0; uint8_t device_id[17]; char pub_topic[200]; char id[50]; uint8_t dw_buffer[255]; uint8_t dw_buffer_len = 0; int mqtt_index = *((int*)argc); printf("dwmqtt_index:%d\r\n", mqtt_index); snprintf(id, 50, "mqtt_modbus_dw_%d", mqtt_index); printf("dwid:%s\r\n", id); mosq_dw_mqtt[mqtt_index] = mosquitto_new(id, true, NULL); mosquitto_connect_callback_set(mosq_dw_mqtt[mqtt_index], mosq_dw_mqtt_connect_callback); mosquitto_disconnect_callback_set(mosq_dw_mqtt[mqtt_index], mosq_dw_mqtt_disconnect_callback); mosquitto_message_callback_set(mosq_dw_mqtt[mqtt_index], mosq_dw_mqtt_message_callback); mosquitto_subscribe_callback_set(mosq_dw_mqtt[mqtt_index], mosq_dw_mqtt_subscribe_callback); mosquitto_username_pw_set(mosq_dw_mqtt[mqtt_index], "leo_mqtt", "leo_mqtt@"); mosquitto_connect(mosq_dw_mqtt[mqtt_index], "localhost", 1883, 600); while (!exit_sig && !quit_sig) { ret = read(vcom_fd[mqtt_index], rx_buffer, 255); if (ret > 0) { rx_buffer_len = ret; printf("info: rx modbus data -->"); for (i = 0; i < rx_buffer_len; i++) printf("0x%x ", rx_buffer[i]); printf("\r\n"); crc = App_Crc16(rx_buffer, rx_buffer_len - 2); if (crc != (*(rx_buffer + rx_buffer_len - 2) << 8 | *(rx_buffer + rx_buffer_len - 1))) { printf("rx modbus crc failed \r\n"); tcflush(vcom_fd[mqtt_index], TCIOFLUSH); continue; } device_addr = rx_buffer[0]; device_funcode = rx_buffer[1]; tx_buffer[0] = device_addr; tx_buffer[1] = device_funcode; switch (device_funcode) { case 0x03: { attr_addr = rx_buffer[2] << 8 | rx_buffer[3]; attr_num = rx_buffer[4] << 8 | rx_buffer[5]; tx_buffer[2] = attr_num * 2; for (j = 0; j < attr_num; j++) { sprintf(attr_addr_str, "reg_%04x", (attr_addr + j)); attr_addr_str[8] = '\0'; pthread_mutex_lock(&mx_db); printf("attr_id:%s --device_addr:%d --appid:%d\r\n", attr_addr_str, device_addr, mqtt_index + 1); select_attr_by_device_addr(&attr_value, device_addr, attr_addr_str, mqtt_index + 1); pthread_mutex_unlock(&mx_db); tx_buffer[3 + 2 * j] = (attr_value >> 8 & 0xff); tx_buffer[3 + 2 * j + 1] = (attr_value >> 0 & 0xff); } crc = App_Crc16(tx_buffer, 3 + attr_num * 2); tx_buffer[3 + 2 * attr_num] = (crc >> 8 & 0xff); tx_buffer[3 + 2 * attr_num + 1] = (crc >> 0 & 0xff); tx_buffer_len = 3 + 2 * attr_num + 2; break; } case 0x06: { attr_addr = rx_buffer[2] << 8 | rx_buffer[3]; attr_value = rx_buffer[4] << 8 | rx_buffer[5]; sprintf(attr_addr_str, "reg_%04x", attr_addr); attr_addr_str[8] = '\0'; sprintf(attr_value_str, "%d", attr_value); attr_value_str[4] = '\0'; tx_buffer[2] = (attr_addr >> 8 & 0xff); tx_buffer[3] = (attr_addr >> 0 & 0xff); tx_buffer[4] = (attr_value >> 8 & 0xff); tx_buffer[5] = (attr_value >> 0 & 0xff); crc = App_Crc16(tx_buffer, 6); tx_buffer[6] = (crc >> 8 & 0xff); tx_buffer[7] = (crc >> 0 & 0xff); tx_buffer_len = 8; if (select_device_id_by_device_addr(device_addr, device_id, mqtt_index) == 1) { printf("attr_addr:%s --- attr-value:%s \r\n", attr_addr_str, attr_value_str); // update sqllite pthread_mutex_lock(&mx_db); update_tab_item_by_device_addr(device_addr, attr_addr_str, attr_value_str, mqtt_index); pthread_mutex_unlock(&mx_db); dwlink_packet_process(device_id, dw_buffer, &dw_buffer_len, attr_addr_str, attr_value_str); // publish dwlink/devices/id/mqtt sprintf(pub_topic, "application/%d/device/%s/command/down", mqtt_index + 1, device_id); mosquitto_publish(mosq_dw_mqtt[mqtt_index], NULL, pub_topic, dw_buffer_len, dw_buffer, 1, false); printf("publish:\r\n"); for (i = 0; i < dw_buffer_len; i++) printf("%02x ", dw_buffer[i]); printf("\r\n"); } break; } default: { tx_buffer[0] = device_addr; tx_buffer[1] = 0x83; tx_buffer[2] = 0x02; tx_buffer_len = 3; break; } } write(vcom_fd[mqtt_index], tx_buffer, tx_buffer_len); printf("info: tx modbus data -->"); for (i = 0; i < tx_buffer_len; i++) printf("0x%x ", tx_buffer[i]); printf("\r\n"); } else if (ret < 0) { continue; } if (mosquitto_loop(mosq_dw_mqtt[mqtt_index], 100, 1)) { mosquitto_reconnect(mosq_dw_mqtt[mqtt_index]); sleep(10); } usleep(500000); } mosquitto_disconnect(mosq_dw_mqtt[mqtt_index]); mosquitto_destroy(mosq_dw_mqtt[mqtt_index]); printf("mqtt_modbus_dwlink_process_thread exit\r\n"); } int vcom_init() { char* dev[] = {"/dev/tty10", "/dev/tty12", "/dev/tty14", "/dev/tty16"}; int i = 0; for (i = 0; i < MAX_VCOM; i++) { vcom_fd[i] = open(dev[i], O_RDWR | O_NOCTTY | O_NDELAY); // open("/dev/tttyS0",O_RDWR | O_NOCTTY); //O_NDELAY if (vcom_fd[i] == -1) { printf("can't open :%s \r\n", dev[i]); return -1; } } printf("vcom fd open success\r\n"); return 1; } int main(int argc, char* argv[]) { pthread_t mqtt_modbus_uplink_tid[MAX_VCOM], mqtt_modbus_dwlink_tid[MAX_VCOM], http_json_tid; int err = 0, i = 0; int thread_index[MAX_VCOM] = {0, 1, 2, 3}; struct sigaction sigact; mosquitto_lib_init(); init_device_db(); pthread_mutex_init(&root_lock, NULL); config_init(); vcom_init(); for (i = 0; i < MAX_VCOM; i++) { printf("mqtt modbus vcom_%d uplink thread\r\n", i); err = pthread_create(&mqtt_modbus_uplink_tid[i], NULL, (void* (*)(void*))mqtt_modbus_uplink_process_thread, (void*)(&thread_index[i])); if (err != 0) { printf("create mqtt modbus vcom_%d uplink thread failed\r\n", i); exit(EXIT_FAILURE); } printf("mqtt modbus vcom_%d dwlink thread\r\n", i); err = pthread_create(&mqtt_modbus_dwlink_tid[i], NULL, (void* (*)(void*))mqtt_modbus_dwlink_process_thread, (void*)(&thread_index[i])); if (err != 0) { printf("create mqtt modbus vcomd_%d dwlink thread failed\r\n", i); exit(EXIT_FAILURE); } } // web printf("http json process thread\r\n"); err = pthread_create(&http_json_tid, NULL, (void* (*)(void*))http_json_process_thread, NULL); if (err != 0) { printf("create http json thread failed\r\n"); exit(EXIT_FAILURE); } sigemptyset(&sigact.sa_mask); sigact.sa_flags = 0; sigact.sa_handler = sig_handler; sigaction(SIGQUIT, &sigact, NULL); sigaction(SIGINT, &sigact, NULL); sigaction(SIGTERM, &sigact, NULL); while (!exit_sig && !quit_sig) { pthread_mutex_lock(&root_lock); get_system_status(); pthread_mutex_unlock(&root_lock); sleep(120); } for (i = 0; i < MAX_VCOM; i++) { pthread_join(mqtt_modbus_uplink_tid[i], NULL); pthread_join(mqtt_modbus_dwlink_tid[i], NULL); } pthread_join(http_json_tid, NULL); if (exit_sig || quit_sig) { // shutdown(net_sock, SHUT_RDWR); } pthread_mutex_destroy(&root_lock); mosquitto_lib_cleanup(); exit(EXIT_SUCCESS); }