#include "cmdclient.h"
#include "mqtt_session.h"
#include "../common.h"
#include <semaphore.h>
#include "../log_module/log_module.h"
mqtt_session_t *session = NULL;
#define CALL_TOPIC "ipc/cmdsvr/call/"APPNAME
static pthread_rwlock_t list_rwlock = PTHREAD_RWLOCK_INITIALIZER;
static list_head_t wirte_list;

static void sem_send_with_times(sem_t *__sem,int times)
{
    for (size_t i = 0; i < times; i++)
    {
        if(sem_post(__sem)==0)
            break;
        proto_syslog(LOG_ERR, "sem_post error :%s",strerror(errno));
    }
}

static int sem_wait_timeout(sem_t *sem,int timeout_s,int timeout_ms) //等待信号量
{	
	struct timeval now;
	struct timespec outtime;
	gettimeofday(&now,NULL);
	outtime.tv_sec = now.tv_sec+timeout_s;
	outtime.tv_nsec = (now.tv_usec+timeout_ms*1000)*1000;
	return sem_timedwait(sem, &outtime);
}

static int cmd_client_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    ems_syslog(LOG_INFO, "cmd client received MQTT topic:%s payload length:%d", mqtt_msg->topic, mqtt_msg->payloadLen);
    ems_syslog(LOG_INFO, "message:%.*s",  mqtt_msg->payloadLen,mqtt_msg->payload);
    char *get_data = calloc(mqtt_msg->payloadLen+1,1);
    memcpy(get_data,mqtt_msg->payload,mqtt_msg->payloadLen);
    cJSON *root = cJSON_Parse(get_data);
    if (root)
    {
        int seq = 0,status = -1,timeout = 0;
        char *resp = NULL;
        cJSON *item = cJSON_GetObjectItem(root, "seq");
        if (item)
        {
            seq = item->valueint;
        }
        item = cJSON_GetObjectItem(root, "status");
        if (item)
        {
            status = item->valueint;
        }
        item = cJSON_GetObjectItem(root, "resp");
        if (item)
        {
            resp = item->valuestring;
        }
        item = cJSON_GetObjectItem(root, "timedout");
        if (item)
        {
            timeout = item->valueint;
        }
        ems_syslog(LOG_INFO,"seq:%d,status:%d,resp:%s,timedout:%d",seq,status,resp,timeout);
        cmd_write_list_t *write_info = NULL;
        pthread_rwlock_wrlock(&list_rwlock);
        list_for_each_entry(write_info, &wirte_list, list)
        {
            if (write_info->cmdInfo->seq == seq)
            {
                sem_send_with_times(write_info->cmdInfo->cmd_info_sem,1);
                write_info->cmdInfo->timeout = timeout;
                if (resp&&(strlen(resp)>0))
                {
                    write_info->cmdInfo->resp = strdup(resp);
                }
                write_info->cmdInfo->status = status;
                // write_info->cmdInfo->resp = strdup(resp);
                // if (-1 == status)
                // {
                //     ems_syslog(LOG_ERR,"system error!");
                //     write_info->cmdInfo->status = -1;
                // }
                // else
                // {
                //     ems_syslog(LOG_NOTICE,"exit status value = [0x%x]\n", status);
            
                //     if (WIFEXITED(status))
                //     {
                //         if (0 == WEXITSTATUS(status))
                //         {
                //             ems_syslog(LOG_NOTICE,"run shell script successfully.\n");
                //             write_info->cmdInfo->status = 0;
                //         }
                //         else
                //         {
                //             ems_syslog(LOG_ERR,"run shell script fail, script exit code: %d\n", WEXITSTATUS(status));
                //             write_info->cmdInfo->status = -1;
                //         }
                //     }
                //     else
                //     {
                //         ems_syslog(LOG_ERR,"exit status = [%d]\n", WEXITSTATUS(status));
                //         write_info->cmdInfo->status = -1;
                //     }
                // }
                write_info->cmdInfo->seq = seq;
            }
        }
        pthread_rwlock_unlock(&list_rwlock);
        cJSON_Delete(root);
    }
    else
    {
        ems_syslog(LOG_ERR,"recv is not json [%s]",get_data);
    }
    free(get_data);
    return 0;
}

int publish_string(char *arg,int timeout,int seq)
{
    int ret = -1;
    char *data_all = calloc(strlen(arg)+256,1);
    sprintf(data_all,"{\"name\":\""APPNAME"\",\"argv\":[\"/bin/sh\", \"-c\",\"%s\"],\"timeout\":%d,\"seq\":%d}",arg,timeout,seq);
    if((ret = mqtt_session_publish(session,CALL_TOPIC,data_all,strlen(data_all)))!=0)
        ems_syslog(LOG_WARNING,"send data error!!");
    free(data_all);
    return ret;
}

int cmd_call(char *arg,int timeout,char **dist_arg,int getreturn)
{
    int ret = -1;
    static int seq = 0;
    seq++;
    ems_syslog(LOG_INFO,"send cmd :[%s]",arg);
    if (getreturn)
    {
        sem_t cmd_sem;
        sem_init(&cmd_sem,0,0);
        CMD_INFO cmdInfo = {.seq = seq,.cmd_info_sem = &cmd_sem};
        cmd_write_list_t write_info = {0};
        write_info.cmdInfo = &cmdInfo;

        pthread_rwlock_wrlock(&list_rwlock);
        list_add_tail(&write_info.list, &wirte_list);
        pthread_rwlock_unlock(&list_rwlock);

        ret = publish_string(arg,timeout,seq);
        
        ret |= sem_wait_timeout(&cmd_sem,timeout,0);
        if (ret == 0)
        {
            pthread_rwlock_wrlock(&list_rwlock);
            if (cmdInfo.timeout == 0)
            {
                ret = cmdInfo.status;
                if (cmdInfo.resp!=NULL)
                {
                    if (dist_arg!=NULL)
                    {
                        if (*dist_arg == NULL)
                        {
                            *dist_arg = cmdInfo.resp;
                            ems_syslog(LOG_INFO,"cmd get resp no free:len [%ld] [%s]",(long int)strlen(*dist_arg),*dist_arg);
                        }
                        else if (*dist_arg!=NULL)
                        {
                            strcpy(*dist_arg,cmdInfo.resp);
                            free(cmdInfo.resp);
                            ems_syslog(LOG_INFO,"cmd get resp:[%s]",*dist_arg);
                        }
                    }
                    else
                        free(cmdInfo.resp);
                }
            }
            else
            {
                ret = -1;
            }
            pthread_rwlock_unlock(&list_rwlock);
        }
        pthread_rwlock_wrlock(&list_rwlock);
        list_del(&write_info.list);
        pthread_rwlock_unlock(&list_rwlock);
    }
    else
    {
        return publish_string(arg,timeout,seq);
    }
    return ret;
}

int cmd_client_init(const char * id)
{
    session =  mqtt_session_new((char *) id, NULL);
    if (session == NULL)
    {
        ems_syslog(LOG_NOTICE,"creat default!!");
        session = NULL;
        return -1;
    }
    ems_syslog(LOG_NOTICE,"cmdclient creat success!!");
    INIT_LIST_HEAD(&wirte_list);
    mqtt_session_set_address(session, "localhost",1883,"","");
    mqtt_session_set_opts(session, 0, 10);

    mqtt_session_set_callbacks(session, cmd_client_recv_msg,NULL);

    char sub_topic[256] = {0};
    sprintf(sub_topic,"ipc/cmdsvr/resp/"APPNAME);
    mqtt_session_subscribe(session, sub_topic);
    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);
    mqtt_session_start(session);
    int times=0;
    while (mqtt_session_get_state(session)!=MQTT_CONNECTED)
    {
        ems_syslog(LOG_NOTICE, "wait cmdclient connect!");
        sleep(1);
        times++;
        if(times>30)
        {
            ems_syslog(LOG_ERR, "cmdclient connect error!");
            break;
        }
    }
    ems_syslog(LOG_NOTICE, "cmdclient connect success!");
    return 0;
}