webconfig/apps/apps.c

#include "common.h"
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <json-c/json.h>
#include <json-c/json_object.h>
#include <json-c/json_util.h>
#include <mosquitto.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/tcp.h>
#include <pthread.h>
#include <signal.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <termios.h>
#include <time.h>
#include <unistd.h>

#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);
}