#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>

#include "ip_alarm.h"
#include "uwsc.h"

#define RTSP_PORT 554
#define IP_ALARM_APP_KEY "5MHgJo6T3eMcvgIu"
#define VIDEO_TIME 120
ip_alarm_var_t gvar = {0};
camera_config_t *ip_ws_cfg = NULL;

static int ipc_load_node_config(ip_alarm_var_t *var, char *file)
{
	char buff[128] = {0};
	int ret 	   = -1;
	int i		   = 0;
	
	char *json_str = NULL;
    json_str = read_file_data(file);
    if (!json_str)
    {
        return -1;
    }

    cJSON *root = cJSON_Parse(json_str);
    if (!root)
    {
        return -1;
    }

    cJSON *ext_data = cJSON_GetObjectItem(root, "ext_data");
    if (!ext_data)
    {
        ret = -1;
        goto __cleanup;
    }

	ip_alarm_config_t *ip_alarm_cfg = calloc(1, sizeof(ip_alarm_config_t));
	ip_ws_cfg = &ip_alarm_cfg->camera_cfg;
	GET_JSON_VALUE_STRING(ext_data, "ws_url", ip_alarm_cfg->camera_cfg.ws_url);
	GET_JSON_VALUE_STRING(ext_data, "ws_host", ip_alarm_cfg->camera_cfg.ws_host);
	GET_JSON_VALUE_STRING(ext_data, "ws_username", ip_alarm_cfg->camera_cfg.ws_username);
	GET_JSON_VALUE_STRING(ext_data, "ws_password", ip_alarm_cfg->camera_cfg.ws_password);	

	cJSON *nodes = cJSON_GetObjectItem(ext_data, "dev_info");
    int size = cJSON_GetArraySize(nodes);
    for (i = 0; i < size ; i++)
    {
        cJSON *node = cJSON_GetArrayItem(nodes, i);
        if (!node)
        {
            ret = i + 1;
            goto __cleanup;
        }

		dev_info_config_t *dev_info_cfg = calloc(1, sizeof(dev_info_config_t));
		GET_JSON_VALUE_STRING(node, "GW_SN", dev_info_cfg->gw_sn);
		GET_JSON_VALUE_STRING(node, "DEV_SN", dev_info_cfg->dev_sn);
		GET_JSON_VALUE_STRING(node, "SERVER_SET", dev_info_cfg->server_set);
		
		dy_syslog(LOG_INFO,"==%d %d gw_sn:%s dev_sn:%s server_set:%s==",size,i,dev_info_cfg->gw_sn,dev_info_cfg->dev_sn,dev_info_cfg->server_set);
		list_add_tail(&dev_info_cfg->list, &var->config_list);
    }
    ret = 0;

__cleanup:
    cJSON_Delete(root);
    return ret;
}

static int ip_alarm_heart_beat(ip_alarm_var_t *var)
{
	TIMER_CONFIRM(var->heartbeat_timer);
    return 0;
}

static int ip_alarm_push_video_timer(ip_alarm_var_t *var)
{
    TIMER_CONFIRM(var->push_video_timer);
    return 0;
}


static void ip_alarm_loop(ip_alarm_var_t *var)
{
    int ret = -1, maxfd;
    fd_set  rset;
    struct timeval timeout;

    while (1)
    {
        SELECT_INIT();
        SELECT_ADD_FD(var->heartbeat_timer);
		SELECT_ADD_FD(var->push_video_timer);
        timeout.tv_usec = 0;
        timeout.tv_sec = 10;
        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            printf("errno %d\n", errno);
            if (errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret == 0)
        {
        }
        else
        {
            if (var->heartbeat_timer > 0 && FD_ISSET(var->heartbeat_timer, &rset))
            {
                FD_CLR(var->heartbeat_timer, &rset);
                ip_alarm_heart_beat(var);
            }
			if (var->push_video_timer > 0 && FD_ISSET(var->push_video_timer, &rset))
            {
                FD_CLR(var->push_video_timer, &rset);
                ip_alarm_push_video_timer(var);
            }
        }
    }
}

static void ip_alarm_subscribe_all(ip_alarm_var_t *var)
{
    ipc_session_t *session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};

    snprintf(topic, TOPIC_MAX_LEN, "ipc/+/%s/device/+/data/%s", IP_ALARM_APP_KEY, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
    ipc_session_subscribe(session, topic);
}

static int ip_alarm_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    ip_alarm_var_t *var = (ip_alarm_var_t *)obj;

    dy_syslog(LOG_DEBUG, " MQTT client: received MQTT topic:%s payload length:%d",
              mqtt_msg->topic, mqtt_msg->payloadLen);
    if (strstr(mqtt_msg->topic, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA))
    {
		return 0;
    }
}

// 建立与内部broker之间的MQTT连接
static int ip_alarm_mqtt_client_init(ip_alarm_var_t *var)
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "client_%s", var->proc_name);
    var->session = ipc_session_new(clientId, (void*)var, IPC_DEFAULT);
    if (var->session == NULL) return -1;

    ipc_session_set_callbacks(var->session, ip_alarm_mqtt_handle_recv_msg, NULL);
    ip_alarm_subscribe_all(var);
    ipc_session_start(var->session);
}

int ip_alarm_external_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    ip_alarm_var_t *var = (ip_alarm_var_t *)obj;
    char *topic = mqtt_msg->topic;

    dy_syslog(LOG_DEBUG, " MQTT client: received MQTT topic:%s payload length:%d",
              mqtt_msg->topic, mqtt_msg->payloadLen);

out:
    return 0;
}

static void ip_alarm_ex_subscribe_all(ip_alarm_var_t *var)
{
    dy_syslog(LOG_INFO, "subscribe all");
    
}


// 建立与broker之间的MQTT连接
int ip_alarm_external_client_init(ip_alarm_var_t *var)
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "IPALARM_%s", var->sn_str);
    var->session_ext = mqtt_session_new(clientId, (void*)var);
    if (var->session_ext == NULL) return -1;

	mqtt_session_set_address(var->session_ext, "mqtt.lnxall.com", 3883, "localuser", "dywl@galaxy");
	
    mqtt_session_set_opts(var->session_ext, DEFAULT_QOS, KEEPALIVEINTERVAL);
    mqtt_session_set_callbacks(var->session_ext, ip_alarm_external_mqtt_handle_recv_msg, NULL);

    mqtt_session_start(var->session_ext);
    ip_alarm_ex_subscribe_all(var);
}

static void ip_alarm_status_update(ip_alarm_config_t *ip_alarm_cfg, int online)
{
	cJSON *jtag = cJSON_CreateObject();

	ip_alarm_cfg->online = online;
	cJSON_AddStringToObject(jtag, "sn", ip_alarm_cfg->camera_cfg.sn);
	cJSON_AddBoolToObject(jtag, "online", online);	
	cJSON_AddNumberToObject(jtag, "last_rcv", time(NULL));
	cJSON_AddNumberToObject(jtag, "login_time", 0);
	
	char *json_str = cJSON_PrintUnformatted(jtag);

	char topic[256] = {0};
	snprintf(topic, sizeof(topic), "ipc/%s/TCP_CLIENT/Evt_NodesStatus", gvar.sn_str);

	ipc_session_publish(gvar.session, topic, json_str, strlen(json_str));
	dy_syslog(LOG_DEBUG, "ip_alarm topic:%s payload:%s", topic, json_str);

	free(json_str);
	cJSON_Delete(jtag);
}

static void stdin_read_cb(struct ev_loop *loop, struct ev_io *w, int revents)
{
    struct uwsc_client *cl = w->data;
    char buf[128] = "";
    int n;

    n = read(w->fd, buf, sizeof(buf));
    if (n > 1) {
        buf[n - 1] = 0;

        if (buf[0] == 'q')
            cl->send_close(cl, UWSC_CLOSE_STATUS_NORMAL, "ByeBye");
        else
            cl->send(cl, buf, strlen(buf) + 1,  UWSC_OP_TEXT);
    }
	else
		usleep(500000);
}

static void uwsc_onopen(struct uwsc_client *cl)
{
    static struct ev_io stdin_watcher;

    dy_syslog(LOG_DEBUG, "onopen");

    stdin_watcher.data = cl;

    ev_io_init(&stdin_watcher, stdin_read_cb, STDIN_FILENO, EV_READ);
    ev_io_start(cl->loop, &stdin_watcher);

	/* Usage of send_ex */
	cJSON *login = cJSON_CreateObject();
	cJSON_AddStringToObject(login, "host", ip_ws_cfg->ws_host);
	cJSON_AddStringToObject(login, "method", "login");
	cJSON_AddStringToObject(login, "username", ip_ws_cfg->ws_username);
	cJSON_AddStringToObject(login, "password", ip_ws_cfg->ws_password);
	char *login_info = cJSON_PrintUnformatted(login);

	cl->send(cl, login_info, strlen(login_info), UWSC_OP_TEXT);
	free(login_info);
	cJSON_Delete(login);

	cJSON *subscribe = cJSON_CreateObject();
	cJSON_AddStringToObject(subscribe, "host", ip_ws_cfg->ws_host);
	cJSON_AddStringToObject(subscribe, "method", "subscribe");
	cJSON_AddStringToObject(subscribe, "username", ip_ws_cfg->ws_username);
	cJSON_AddStringToObject(subscribe, "password", ip_ws_cfg->ws_password);
	char *subscribe_info = cJSON_PrintUnformatted(subscribe);
	
	sleep(1);
	cl->send(cl, subscribe_info, strlen(subscribe_info), UWSC_OP_TEXT);	
	free(subscribe_info);
	cJSON_Delete(subscribe);

    //cl->send(cl, "{'host':'118.31.12.250', 'method':'login','username':'djx','password':'djx123'}", strlen("{'host':'118.31.12.250', 'method':'login','username':'djx','password':'djx123'}"), UWSC_OP_TEXT);
	//sleep(1);
	//cl->send(cl, "{'host':'118.31.12.250','method':'subscribe','username':'djx','password':'djx123'}", strlen("{'host':'118.31.12.250','method':'subscribe','username':'djx','password':'djx123'}"), UWSC_OP_TEXT);
}

static void uwsc_onmessage(struct uwsc_client *cl,	void *data, size_t len, bool binary)
{
    if (binary) {
        size_t i;
        uint8_t *p = data;
		dy_syslog(LOG_DEBUG, "Recv:");
        for (i = 0; i < len; i++) {
            dy_syslog(LOG_DEBUG, "%02hhX ", p[i]);
            if (i % 16 == 0 && i > 0)
                puts("");
        }
        puts("");
    }
	else
	{
		cJSON *root = cJSON_Parse(data);
	    if (root)
	    {
	        char buf[512] = {0};
			char method[64] = {0};

			GET_JSON_VALUE_STRING(root, "method", method);
			GET_JSON_VALUE_STRING(root, "ExtenInfo", buf);
			
			//if(!strstr(method, "online") && !strstr(method, "offline"))
			dy_syslog(LOG_DEBUG, "Recv:[%.*s]", (int)len, (char *)data);
			
			//ip_alarm_config_t *ip_alarm_cfg = NULL;
			dev_info_config_t *dev_info_cfg = NULL;
			list_for_each_entry(dev_info_cfg, &gvar.config_list, list)
			{
				dy_syslog(LOG_DEBUG, "buf:%s sn:%s", buf, dev_info_cfg->dev_sn);
				if(strstr(buf, dev_info_cfg->dev_sn))
				{
					dy_syslog(LOG_DEBUG, "Recv:[%.*s]", (int)len, (char *)data);
					dy_syslog(LOG_DEBUG, "sn:%s", dev_info_cfg->dev_sn);
					if(strstr(method, "ring") || strstr(method, "handup") || 
						strstr(method, "online") || strstr(method, "offline"))
					{
						//发布MQTT消息
				        char *data_tmp = NULL;
						time_t now = time(NULL);
						
				        cJSON *root = cJSON_CreateObject();
						cJSON_AddStringToObject(root, "sn", dev_info_cfg->dev_sn);
				        cJSON_AddNumberToObject(root, "mi", 0);
				        cJSON_AddNumberToObject(root, "timestamp", now);
				        cJSON_AddStringToObject(root, "identifier", method);
						
						data_tmp = cJSON_Print(root);
						dy_syslog(LOG_DEBUG, "data_tmp:%s", data_tmp);
						mqtt_session_publish(gvar.session_ext, dev_info_cfg->server_set, data_tmp, strlen(data_tmp));
						
						free(data_tmp);
					#if 0
						if(ip_alarm_cfg->online == false)
						{
							ip_alarm_status_update(ip_alarm_cfg, 1);
						}
						
						pp_real_time_data_t real_data = {0};
						strcpy(real_data.sn, ip_alarm_cfg->camera_cfg.sn);
						strcpy(real_data.src_identifier, "StatusUpdate");
						
						cJSON *jtag = cJSON_CreateObject();
						cJSON_AddStringToObject(jtag, "status", method);
						char *tag = cJSON_PrintUnformatted(jtag);			
						ipc_data_reply(&gvar, &real_data, tag, strlen(tag));
						
						free(tag);
						cJSON_Delete(jtag);

						if(strstr(method, "ring"))
						{
							time_t now = time(NULL);
							if(now - ip_alarm_cfg->last_push > 120)
							{
								ip_alarm_cfg->last_push = now;
								ip_alarm_cfg->push_video_flag = 1;
							
								char *record = calloc(512, 1);

								snprintf(ip_alarm_cfg->video_file, sizeof(ip_alarm_cfg->video_file), "%d.mp4", ip_alarm_cfg->last_push);
								snprintf(record, 512, "ffmpeg -rtsp_transport tcp -nostdin -i \"%s\" -t %d -f mp4 /tmp/%s 2>&1 | logger -t 'ffmpeg' & ", ip_alarm_cfg->camera_cfg.rtsp_url,VIDEO_TIME,ip_alarm_cfg->video_file);

								if(record && strlen(record))
								{
									dy_syslog(LOG_DEBUG, "record:%s", record);
									system(record);
									free(record);
								}
							}
						}
					#endif
					#if 0
						ip_alarm_data_t *ip_alarm_data = calloc(1, sizeof(ip_alarm_data_t));
						ip_alarm_data->len = len;
						ip_alarm_data->data = calloc(1,len+1);
						memcpy(ip_alarm_data->data, data, len);
						
						dy_syslog(LOG_INFO,"==Recv len:%d data:%s", len, ip_alarm_data->data);
						list_add_tail(&ip_alarm_data->list, &gvar.data_list);
					#endif
					}
					else if(strstr(method, "online"))
					{
						//ip_alarm_status_update(ip_alarm_cfg, 1);
					}
					else if(strstr(method, "offline"))
					{
						//ip_alarm_status_update(ip_alarm_cfg, 0);
					}
					break;
				}							
			}
			cJSON_Delete(root);
	    }		
    }
}

static void uwsc_onerror(struct uwsc_client *cl, int err, const char *msg)
{
    dy_syslog(LOG_ERR, "onerror:%d: %s", err, msg);
    ev_break(cl->loop, EVBREAK_ALL);
	dy_syslog(LOG_INFO, "ip_alarm connect fail exit...");
	exit(1);
}

static void uwsc_onclose(struct uwsc_client *cl, int code, const char *reason)
{
    dy_syslog(LOG_ERR, "onclose:%d: %s", code, reason);
    ev_break(cl->loop, EVBREAK_ALL);
}

static void signal_cb(struct ev_loop *loop, ev_signal *w, int revents)
{
    if (w->signum == SIGINT) {
        ev_break(loop, EVBREAK_ALL);
        dy_syslog(LOG_INFO, "Normal quit");
    }
}

void *uwsc_loop(void *param)
{
	//camera_config_t *camera_cfg = (camera_config_t *)param;
	const char *url = ip_ws_cfg->ws_url;//"wss://118.31.12.250/DisWebSocket/websocket";
    struct ev_loop *loop = EV_DEFAULT;
    struct ev_signal signal_watcher;
	int ping_interval = 10;	/* second */
    struct uwsc_client *cl;
	int opt;
	
	dy_syslog(LOG_INFO, "Libuwsc: %s", UWSC_VERSION_STRING);

	while(1)
	{
	    cl = uwsc_new(loop, url, ping_interval, NULL);
		dy_syslog(LOG_INFO, "cl %p", cl);
	    if (cl)
	    {
			dy_syslog(LOG_INFO, "Start connect...");

		    cl->onopen = uwsc_onopen;
		    cl->onmessage = uwsc_onmessage;
		    cl->onerror = uwsc_onerror;
		    cl->onclose = uwsc_onclose;

		    //ev_signal_init(&signal_watcher, signal_cb, SIGINT);
		    //ev_signal_start(loop, &signal_watcher);
		    ev_run(loop, 0);

		    free(cl);
	    }
		dy_syslog(LOG_INFO, "ip_alarm connect fail exit...");
		exit(1);
	}
}

static int ip_alarm_init(ip_alarm_var_t *var)
{
	int i;
	int start = 0;
	
	INIT_LIST_HEAD(&var->config_list);
	INIT_LIST_HEAD(&var->data_list);
	
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);

	system("killall -15 ffmpeg");
	system("killall -15 ffmpeg");

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }
	if (load_templates_cfg(&var->template_table, TEMPLATES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load template cfg fail");
    }
	
    if (access(IP_CAMERA_CFG, F_OK) != -1)
    {
        ipc_load_node_config(var, IP_CAMERA_CFG);
    }

	ip_alarm_mqtt_client_init(var);
	ip_alarm_external_client_init(var);

	if(!ip_ws_cfg || !strlen(ip_ws_cfg->ws_url))
	{
		while(1)
		{
			dy_syslog(LOG_INFO, "Libuwsc: %s ip_ws_cfg is NULL or ws_url len is 0 !!!", UWSC_VERSION_STRING);
			sleep(30);
		}
	}
    var->heartbeat_timer = my_timer_create();
    if (var->heartbeat_timer > 0)
    {
    	my_timer_set(var->heartbeat_timer, 3, 300000);
    }

	var->push_video_timer = my_timer_create();
    if (var->push_video_timer > 0)
    {
    	my_timer_set(var->push_video_timer, 3, 3000);
    }

	//update_depend_package_info(var);

    dy_syslog(LOG_INFO, "init done, board SN:%s", var->sn_str);

    return 0;
}

int main(int argc, char *argv[])
{
    ip_alarm_var_t *var = &gvar;
	pthread_t thread_uwsc;

    memset(var, 0, sizeof(ip_alarm_var_t));

    ip_alarm_init(var);	
	pthread_create(&thread_uwsc, 0, uwsc_loop, NULL);
    ip_alarm_loop(var);

    return 0;
}