#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_soundbox.h"
#define DATA_PATH "/app/data/"
#define IP_SOUNDBOX_APP_KEY "7FQwy0CyVqW3izpA"

ip_soundbox_var_t gvar = {0};
gw_port_e ports[] = {TCP_CLIENT};

static int get_rt_data_from_json_str(pp_real_time_data_t *real_data, char *json_str)
{
    int ret = 0;

    cJSON *root = cJSON_Parse(json_str);
    if (root)
    {
        char *payload = NULL;
        char *buf = NULL;

        buf = (char *)malloc(MAXBUF);
        if (buf == NULL)
        {
            dy_syslog(LOG_ERR, "malloc failed");
            ret = -1;
            goto out;
        }

        GET_JSON_VALUE_STRING(root, "port", buf);
        real_data->port = port_char2enum(buf);
        GET_JSON_VALUE_INT(root, "instruction_code", real_data->instruction_code);
        GET_JSON_VALUE_INT(root, "mi", real_data->mi);
        GET_JSON_VALUE_INT(root, "ts", real_data->ts);
        GET_JSON_VALUE_INT(root, "len", real_data->len);
        GET_JSON_VALUE_INT(root, "sub_term_addr_len", real_data->sub_term_addr_len);
        GET_JSON_VALUE_INT(root, "start_addr", real_data->start_addr);
        GET_JSON_VALUE_STRING(root, "sub_template_id", real_data->sub_template_id);
        GET_JSON_VALUE_STRING(root, "src_identifier", real_data->src_identifier);
        GET_JSON_VALUE_STRING(root, "sn", real_data->sn);
        GET_JSON_VALUE_STRING(root, "dtu_sn", real_data->dtu_sn);

        GET_JSON_VALUE_STRING(root, "data_b64", buf);
		dy_syslog(LOG_DEBUG, "buf:%s, real_data->len:%d", buf, real_data->len);
        real_data->data = calloc(real_data->len+1, 1);
        if (real_data->data == NULL)
        {
            dy_syslog(LOG_ERR, "malloc fail len:%d", real_data->len);
            free(buf);
            goto out;
        }
        memcpy(real_data->data, buf, real_data->len);
		dy_syslog(LOG_DEBUG, "real_data->data:%s, real_data->len:%d", real_data->data, real_data->len);
		free(buf);
    }
    else
    {
        dy_syslog(LOG_ERR, "json parse error");
        ret = -1;
    }

out:
    cJSON_Delete(root);

    return ret;
}

static int get_lan_port_ip(char *lan_ip_addr)
{
    int ret = -1;
    char buff[255] = {0};
    char *string = NULL;
    snprintf(buff, sizeof(buff), "ubus call network.interface.lan status | grep \"address\" | grep -oE '[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}'; ");
    ret = system_reply(buff, &string);
    if (ret == 0)
    {
        if (!string)return 0;
		//strcpy(wan_ip_addr, string);
		//dy_syslog(LOG_DEBUG, "---string %s", string);
		if(strlen(string))
		{
			unsigned char ip_buf[4] = {0};
			int cnt = sscanf(string, "%d.%d.%d.%d", &ip_buf[0], &ip_buf[1], &ip_buf[2], &ip_buf[3]);
			snprintf(lan_ip_addr, 32, "%d.%d.%d.%d", ip_buf[0], ip_buf[1], ip_buf[2], ip_buf[3]);
			//dy_syslog(LOG_DEBUG, "--cnt %d lan_ip_addr:%s len:%d", cnt,lan_ip_addr,strlen(lan_ip_addr));
		}
        free(string);
		return 1;
    }
    return -1;
}

static char *generate_reply_data(int result)
{
    cJSON *node = cJSON_CreateObject();
    time_t now = time(NULL);

    cJSON_AddNumberToObject(node, "result", result);

    char *data = cJSON_Print(node);
    cJSON_Delete(node);

    return data;
}

static int ip_soundbox_heart_beat(ip_soundbox_var_t *var)
{
    int ret;

	TIMER_CONFIRM(var->heartbeat_timer);
    return 0;
}

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

    while (1)
    {
        SELECT_INIT();
        SELECT_ADD_FD(var->heartbeat_timer);
        timeout.tv_usec = 0;
        timeout.tv_sec = 10;
        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            dy_syslog(LOG_ERR, "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_soundbox_heart_beat(var);
            }
        }
    }
}

static void ip_soundbox_subscribe_all(ip_soundbox_var_t *var)
{
    ipc_session_t *ipc_session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};
	int i = 0;
	
	for (i = 0; i < ARRAY_SIZE(ports); i++)
    {
        snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/+/data/%s", var->sn_str, IP_SOUNDBOX_APP_KEY, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
        ipc_session_subscribe(ipc_session, topic);
    }
}

static void ip_soundbox_data_reply(ip_soundbox_var_t *var, pp_real_time_data_t *real_data, char *result_data, int len)
{
	tag_table_t ptag = {0};
	tag_table_t ptag_post = {0};
					
	strcpy(ptag.sn, real_data->sn);
	ptag.mi = real_data->mi;		
	strncpy(ptag.identifier, real_data->src_identifier, strlen(real_data->src_identifier));
	ptag.report_period = 0;
	ptag.data_type = DATA_TYPE_SERVICE;
	ptag.tag_node = calloc((len)+1, 1);
	memcpy(ptag.tag_node, result_data, len);
	ptag.time = time(NULL);

	memcpy(&ptag_post, &ptag, sizeof(tag_table_t));
	strcpy(ptag_post.identifier, "post");
	ptag_post.data_type = DATA_TYPE_PROPERTY;
	ptag_post.tag_node = calloc((len)+1, 1);
	memcpy(ptag_post.tag_node, result_data, len);

	dy_syslog(LOG_DEBUG, "parse result_data:%s post:%s", ptag.tag_node, ptag_post.tag_node);
    if (ptag.tag_node != NULL)
    {
		ptag.port = real_data->port;
        pp_send_tag_data(var->session, &ptag);
        free(ptag.tag_node);
    }
    if (ptag_post.tag_node != NULL)
    {
		ptag_post.port = real_data->port;
        pp_send_tag_data(var->session, &ptag_post);
        free(ptag_post.tag_node);
    }
}

static int ip_soundbox_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    ip_soundbox_var_t *var = (ip_soundbox_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))
    {
    	int ret = 0;
    	pp_real_time_data_t real_data = {0};

	    ret = get_rt_data_from_json_str(&real_data, mqtt_msg->payload);
	    if (ret < 0)
	    {
	        dy_syslog(LOG_ERR, "parse real data structure failed");
	    }
		#ifdef DATA_SERVICE
		if(strcmp(real_data.src_identifier, "DownloadFile") == 0)
		{
			char url[256] = {0};
			cJSON *root = cJSON_Parse(real_data.data);
		    if (root)
		    {
		        GET_JSON_VALUE_STRING(root, "url", url);
				dy_syslog(LOG_INFO, "====url:%s", url);
				ret = download_file(DATA_PATH, url, 10);
		        if (ret != 0)
		        {
		            dy_syslog(LOG_INFO, "download file fail");
		        }
				char *data = generate_reply_data(ret);				
				ip_soundbox_data_reply(var, &real_data, data, strlen(data));
				free(data);
		    }
		}
		#endif
		client_info_t *client_info = NULL;
		list_for_each_entry(client_info, &var->client_list, list)
		{
			if(strcmp(client_info->sn, real_data.sn) == 0)
			{
			#ifdef DATA_SERVICE
				if(strcmp(real_data.src_identifier, "Stop") == 0)
				{
					dy_session_close_client(&var->socket_server, client_info->socket_fd);
				}
				else if(strcmp(real_data.src_identifier, "PlayMusic") == 0)
				{
					//dy_syslog(LOG_DEBUG, "IP SoundBox Auto Close...");
					//dy_session_close_client(&var->socket_server, client_info->socket_fd);
				}
			#else
				if(strcmp(real_data.src_identifier, "PlayMusic") == 0)
				{
					//store the identifier
				    {
				        char key_str[64] = {0};
				        regulate_cmd_backup_t fields;

				        strcpy(fields.src_identifier, real_data.src_identifier);
				        strcpy(fields.sn, real_data.sn);
				        fields.mi = real_data.mi;

				        sprintf(key_str, "%s", IP_SOUNDBOX_APP_KEY);
				        var->identifier_backup.insert(&var->identifier_backup, key_str, &fields, sizeof(fields));
				    }
					cJSON *root = cJSON_Parse(real_data.data);
				    if (root)
				    {
				    	char buf[128] = {0};
						int port = 0;
						
				    	GET_JSON_VALUE_STRING(root, "dataserver", buf);
						GET_JSON_VALUE_INT(root, "dataserverport", port);

						if(strlen(buf) < 4 || port == 0)
						{
							char lan[64] = {0};
							
							get_lan_port_ip(lan);
							cJSON_AddStringToObject(root, "dataserver", lan);
							cJSON_AddNumberToObject(root, "dataserverport", 8899);
							char *data = cJSON_PrintUnformatted(root);
							dy_syslog(LOG_DEBUG, "lan:%s data:%s", lan,data);
							dy_socket_server_session_snd(&var->socket_server, client_info->socket_fd, data, strlen(data), NULL);
							free(data);
						}
						else
							dy_socket_server_session_snd(&var->socket_server, client_info->socket_fd, real_data.data, real_data.len, NULL);
						
						cJSON_Delete(root);
				    }
				}
				else if(strcmp(real_data.src_identifier, "GetStatus") == 0)
				{
					ip_soundbox_data_reply(var, &real_data, client_info->status, strlen(client_info->status));
				}
			#endif
				break;
			}
		}

		#ifdef DATA_SERVICE
		if(strcmp(real_data.src_identifier, "Stop") == 0)
		{
			sleep(6);
			char *data = generate_reply_data(0);				
			ip_soundbox_data_reply(var, &real_data, data, strlen(data));
			free(data);
		}
		#endif
		
		if (real_data.data)
	    {
	        free(real_data.data);
	    }
    }
}

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

    snprintf(clientId, MAX_CLIENT_ID_LEN, "%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_soundbox_mqtt_handle_recv_msg, NULL);
    ip_soundbox_subscribe_all(var);
    ipc_session_start(var->session);
}

static int socket_client_up_down(struct socket_session_s *session, int up, char *hostname, int socket_fd)
{
	ip_soundbox_var_t *var = &gvar;	

	dy_syslog(LOG_DEBUG, "up %d hostname %s socket_fd %d", up, hostname, socket_fd);
	
	if(up)
	{
		client_info_t *client_info = calloc(1, sizeof(client_info_t));
		strncpy(client_info->hostname, hostname, 64);
		client_info->socket_fd = socket_fd;

		int i;
		for(i=0; i<var->nodes_cfg_table->node_cnt; i++)
		{			
			if(strcmp(var->nodes_cfg_table->node[i].tcp_ip_addr, client_info->hostname) == 0)
			{
				strcpy(client_info->sn, var->nodes_cfg_table->node[i].sn);
				//dy_syslog(LOG_DEBUG, "client_info->sn %s up %d hostname %s socket_fd %d tcp_ip_addr %s", client_info->sn, up, hostname, socket_fd, var->nodes_cfg_table->node[i].tcp_ip_addr);
				break;
			}
		}

		dy_syslog(LOG_DEBUG, "UP sn %s up %d hostname %s socket_fd %d tcp_ip_addr %s", var->nodes_cfg_table->node[i].sn, up, hostname, socket_fd, var->nodes_cfg_table->node[i].tcp_ip_addr);
		list_add_tail(&client_info->list, &var->client_list);
	}
	else
	{
		client_info_t *client_info = NULL;
		list_for_each_entry(client_info, &var->client_list, list)
		{
			if(socket_fd == client_info->socket_fd)
			{
				dy_syslog(LOG_INFO,"DOWN hostname:%s socket_fd:%d",client_info->hostname,client_info->socket_fd);
				list_del(&client_info->list);
				break;
			}
		}
	}
}

static int handle_data_send_status(socket_server_session_t *session, int client_fd, int status, char *user_param)
{
	char topic[TOPIC_MAX_LEN] = {0};
	char cmd[64] = {0};
	ip_soundbox_var_t *var = &gvar;	
    connect_config_t *connect_cfg = container_of(session, connect_config_t, socket_server);
	
#ifdef DATA_SERVICE
    client_info_t *tmp_client = NULL;
	list_for_each_entry(tmp_client, &var->client_list, list)
	{
		if(tmp_client->socket_fd == client_fd && status == 0 && strstr(user_param, "get"))
		{
			dy_syslog(LOG_DEBUG, "IP SoundBox play finished auto close...");
			dy_session_close_client(&var->socket_server, tmp_client->socket_fd);
			break;
		}
	}
#endif
}

static int socket_handle_rcv_msg(socket_server_session_t *session, int client_fd, char *buf, int len)
{
	char topic[TOPIC_MAX_LEN] = {0};
	char cmd[64] = {0};
	ip_soundbox_var_t *var = &gvar;	
    connect_config_t *connect_cfg = container_of(session, connect_config_t, socket_server);

    client_info_t *tmp_client = NULL;
	list_for_each_entry(tmp_client, &var->client_list, list)
	{
		if(tmp_client->socket_fd == client_fd)
		{
			dy_syslog(LOG_DEBUG, "data %s len %d			网关<-[%s]		", buf, len, tmp_client->hostname);
			break;
		}
	}

#ifdef DATA_SERVICE
	int lens = 0;

	client_info_t *client_info = NULL;
	list_for_each_entry(client_info, &var->client_list, list)
	{
		dy_syslog(LOG_INFO, "FIND socket_fd(%d:%d)", client_info->socket_fd, client_fd);
		if(client_info->socket_fd == client_fd)
		{
			char tmp[128] = {0};
			char *path = NULL;
			
			memcpy(tmp, buf, len);
			path = strstr(tmp, "get");
			if(path)
			{
				snprintf(client_info->file_name, sizeof(client_info->file_name), "%s%s", DATA_PATH, path+4);
				
				char *data = read_file_data_and_lens(client_info->file_name, &lens);
			    if (data == NULL)
			    {
			        dy_syslog(LOG_ERR, "read cfg file %s error", client_info->file_name);
					dy_syslog(LOG_DEBUG, "IP SoundBox auto close...");
					dy_session_close_client(&var->socket_server, client_info->socket_fd);
			        return 0;
			    }

				dy_syslog(LOG_DEBUG, "IP SoundBox send sound...");
				dy_socket_server_session_snd(&connect_cfg->socket_server, client_fd, data, lens, tmp);
				client_info->start = 1;
			}
			break;
		}
	}
#else
	cJSON *node = cJSON_Parse(buf);
    if (!node)
    {
        dy_syslog(LOG_WARNING, "json rsData %s error!!!", buf);
        return 0;
    }
    GET_JSON_VALUE_STRING(node, "command", cmd);

	if(strcmp(cmd, "register") == 0)
	{
		cJSON *rs_data = cJSON_CreateObject();
		cJSON_AddStringToObject(rs_data, "command", "register");
		cJSON_AddNumberToObject(rs_data, "result", 200);

		char *data_tmp = cJSON_Print(rs_data);

		dy_socket_server_session_snd(&connect_cfg->socket_server, client_fd, data_tmp, strlen(data_tmp), NULL);

		// 回收资源
	    free(data_tmp);
	    cJSON_Delete(rs_data);
		dy_syslog(LOG_DEBUG, "IP SoundBox register...");
	}
	else if(strcmp(cmd, "start") == 0)
	{
		int result = 0;
		pp_real_time_data_t real_data = {0};
		regulate_cmd_backup_t *fields = NULL;
		
		GET_JSON_VALUE_INT(node, "result", result);

		//restore the identifier
	    {
	        char key_str[64] = {0};
	        data_info_t *data_info = NULL;

	        sprintf(key_str, "%s", IP_SOUNDBOX_APP_KEY);
	        data_info = var->identifier_backup.get(&var->identifier_backup, key_str);
	        if (data_info)
	        {
	            fields = (regulate_cmd_backup_t *)data_info->data;
	        }
	    }
	    if (fields != NULL)
	    {
	        real_data.mi = fields->mi;
	        strcpy(real_data.sn, fields->sn);
	        strncpy(real_data.src_identifier, fields->src_identifier, sizeof(real_data.src_identifier));
	    }

		char *data = generate_reply_data(result == 200 ? 0 : result);			
		ip_soundbox_data_reply(var, &real_data, data, strlen(data));
		free(data);
		dy_syslog(LOG_DEBUG, "IP SoundBox play finished...");
	}
	else if(strcmp(cmd, "status") == 0)
	{
		if(len < sizeof(tmp_client->status) && tmp_client)
		{
			memset(tmp_client->status, 0, sizeof(tmp_client->status));
			memcpy(tmp_client->status, buf, len);
		}
	}
		
#endif
}

static int ip_soundbox_load_conf(ip_soundbox_var_t *var, char *file)
{
	return 0;
}

static int ip_soundbox_init(ip_soundbox_var_t *var)
{	
	INIT_LIST_HEAD(&var->client_list);
	kv_array_init(&var->identifier_backup, 32);
	
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }

    ip_soundbox_mqtt_client_init(var);
	
    //if (access(COMMON_SOCKET_CFG, F_OK) != -1)
    {
        ip_soundbox_load_conf(var, COMMON_SOCKET_CFG);
		var->socket_server.socket_server_port = TCP_SERVER_PORT;
		var->socket_server.type = SOCKET_TCP;
		var->socket_server.handle_recv_msg = socket_handle_rcv_msg;
		var->socket_server.handle_client_up_down = socket_client_up_down;
		var->socket_server.handle_data_send_status = handle_data_send_status;
	
		dy_syslog(LOG_INFO,"==port:%d protocol_type:%d==",var->socket_server.socket_server_port,var->socket_server.type);
		dy_socket_server_session_init(&var->socket_server);
    }

    var->heartbeat_timer = my_timer_create();
    if (var->heartbeat_timer > 0)
    {
    	if(var->node_cnt)
			my_timer_set(var->heartbeat_timer, 3, 1000);
		else
        	my_timer_set(var->heartbeat_timer, 3, 10000);
    }

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

    return 0;
}

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

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

    ip_soundbox_init(var);
    ip_soundbox_loop(var);

    return 0;
}