//=========kk=============
#include <pthread.h>
#include "com_api.h"
#include "kk_log.h"

#define APP2MID         "ipc:///tmp/app2mid.ipc"
#define APP2MID_PUBSUB  "ipc:///tmp/app2mid_pubsub.ipc"
#define PLAT2MID        "ipc:///tmp/plat2mid.ipc"
//#define GW2CCU_PIPE     "tcp://%s:5555"
//#define GW2CCU_PUBSUB   "tcp://%s:5557"
#define GW2CCU_PIPE     "tcp://%s:35567"
#define GW2CCU_PUBSUB   "tcp://%s:35568"

#define MAGIC           "magic12"
#define MAGIC_ACK       "magic12ack"
#define FILTERSTR "|"

typedef struct {
    int n;//nanomsg socket
    int s;//nanomsg recieve fd
}nanomsg_info_t;

typedef struct {
	nanomsg_info_t ba;
	nanomsg_info_t ab;
	char subscrStr[20];
	struct ev_io watcher;
	ipc_cb* cb;
	ipc_type type;
	int isconnect;
	int sendErrCnt;
}Bloop_ctrl_t;


static Bloop_ctrl_t Bloop_ctrl;
static Bloop_ctrl_t Mloop_ctrl;

static struct ev_loop* gloop = NULL;
static pthread_t g_pTh = 0;


static char* _parse_data_by_subscribe(char* data, int len, uint32_t *outLen, void** chlMark){
    char* foundStr = NULL;
    if (data == NULL|| len <= 0 ){
        return data;
    }

    if ( (foundStr= strstr(data, FILTERSTR)) != NULL){

        int ind = foundStr - data;
        data[ind] = 0;
        *chlMark = data;
        *outLen = (len - (foundStr - data) -1);
        return  (foundStr + 1);
    }
    
    *outLen = len;
    return data;
    
}

static void watcher_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
{

    //INFO_PRINT("watcher_cb !! \n");
    //void *user_data = ev_userdata(loop);
    Bloop_ctrl_t *loop_ctrl = (Bloop_ctrl_t *)w->data;
    uint8_t *dat = NULL;
    uint8_t *validDat = NULL;
    uint8_t *chlMark = NULL;
    uint32_t bytes =0;
    uint32_t validLen =0;
    char *dat_bak = NULL;	
    if (loop_ctrl->type == IPC_PLAT2MID ||loop_ctrl->type == IPC_APP2MID){
        bytes = nn_recv(loop_ctrl->ab.n, &dat, NN_MSG, NN_DONTWAIT);
    }else{
        bytes = nn_recv(loop_ctrl->ba.n, &dat, NN_MSG, NN_DONTWAIT);
    }
    if (bytes <= 0 || dat == NULL) {
        //ERROR_PRINT(" recived data is null or len is 0 \n");
        //ERROR_PRINT("nn_recv failed with error code %d, %s \n", nn_errno(), nn_strerror(nn_errno ()));
        return;
    }
    //INFO_PRINT("watcher_cb:-------------------------\n");
	dat_bak = malloc(bytes+1);
	if(dat_bak != NULL){
		memset(dat_bak,0x0,bytes+1);
		memcpy(dat_bak,dat,bytes);
		cJSON_Minify((char *)dat_bak);
    	INFO_PRINT("watcher_cb:%s\n", (char*)dat_bak);	
		free(dat_bak);
		dat_bak = NULL;
	}
    loop_ctrl->isconnect = 1;
     //if sub, need filter sbuscribe str
    if (IPC_PLAT2MID == loop_ctrl->type || IPC_MID2PLAT == loop_ctrl->type){
        validDat = _parse_data_by_subscribe((char*)dat, bytes, &validLen, &chlMark);
    }else{
        validDat = dat;
        validLen = bytes;
    }

    if (strncmp((char*)validDat,MAGIC, strlen(MAGIC)) == 0){
        if (strncmp((char*)validDat,MAGIC_ACK, strlen(MAGIC_ACK)) == 0){
            
        }else{
            kk_ipc_send_ex(loop_ctrl->type, MAGIC_ACK, strlen(MAGIC_ACK)+1, chlMark);
        }

        loop_ctrl->isconnect =1;
        nn_freemsg(dat);
        return;
    }else if (IPC_MID2PLAT == loop_ctrl->type){//
        loop_ctrl->isconnect =0;
    }
	if (loop_ctrl->cb != NULL){
		loop_ctrl->cb((void *)validDat, validLen, chlMark);
	}
    nn_freemsg(dat);
}


void __loop_init(Bloop_ctrl_t *loop_ctrl, struct ev_loop* loop)
{
    
    loop_ctrl->watcher.data = loop_ctrl;
    if (loop_ctrl->type == IPC_PLAT2MID || loop_ctrl->type == IPC_APP2MID){
        ev_io_init (&(loop_ctrl->watcher), watcher_cb, loop_ctrl->ab.s, EV_READ);
    }else{
        ev_io_init (&(loop_ctrl->watcher), watcher_cb, loop_ctrl->ba.s, EV_READ);
    }
   
    ev_io_start (loop, &(loop_ctrl->watcher));
}


int __nanomsg_init(Bloop_ctrl_t *loop_ctrl, ipc_type type,char* chlMark, char* ip)
{
	loop_ctrl->ba.n = -1;
	loop_ctrl->ba.s = -1;
	loop_ctrl->ab.n = -1;
	loop_ctrl->ab.s = -1;
	memset(loop_ctrl->subscrStr, 0, sizeof(loop_ctrl->subscrStr));
	if (chlMark != NULL){
		if (strlen(chlMark) > sizeof(loop_ctrl->subscrStr) +1){
			ERROR_PRINT("__nanomsg_init chlMark is too long, need less than %d  \r\n", 
							sizeof(loop_ctrl->subscrStr));
			return -1;
		}
		memcpy(loop_ctrl->subscrStr, chlMark, strlen(chlMark));
	}
	switch (type) {
		case IPC_APP2MID:{
			loop_ctrl->ba.n = nn_socket(AF_SP, NN_PUSH);
			if (loop_ctrl->ba.n < 0) {
				ERROR_PRINT("__nanomsg_init loop_ctrl->ba.n=%d \r\n",loop_ctrl->ba.n);
				return -1;
			}
			loop_ctrl->ab.n = nn_socket(AF_SP, NN_SUB);
			if (loop_ctrl->ab.n < 0) {
				ERROR_PRINT("loop_ctrl->ab.n =%d \r\n",loop_ctrl->ab.n);
				return -1;
			}
			if (nn_connect(loop_ctrl->ba.n, APP2MID) < 0) {
				return -1;
			}
			//订阅
			if (nn_setsockopt(loop_ctrl->ab.n, NN_SUB, NN_SUB_SUBSCRIBE, "", 0) < 0) {
				ERROR_PRINT("nn_setsockopt failed ");
				return -1;
			}
			if (nn_connect(loop_ctrl->ab.n, APP2MID_PUBSUB) < 0) {//sub
				printf("[%s][%d]\n",__FUNCTION__,__LINE__);
				return -1;
			}
		}
        break;
        case IPC_MID2APP: {
            loop_ctrl->ba.n = nn_socket(AF_SP, NN_PULL);
            if (loop_ctrl->ba.n < 0) {
                ERROR_PRINT("__nanomsg_init loop_ctrl->ba.n=%d \r\n",loop_ctrl->ba.n);
                return -1;
            }
            loop_ctrl->ab.n = nn_socket(AF_SP, NN_PUB);
            if (loop_ctrl->ab.n < 0) {
                ERROR_PRINT("__nanomsg_init loop_ctrl->ab.n =%d \r\n",loop_ctrl->ab.n);
                return -1;
            }            
            if (nn_bind(loop_ctrl->ba.n, APP2MID) < 0) {
                ERROR_PRINT("__nanomsg_init loop_ctrl->ba.n nn_bind error \r\n");
		        return -1;
		    }

		    if (nn_bind(loop_ctrl->ab.n, APP2MID_PUBSUB) < 0) {//pub
                ERROR_PRINT("__nanomsg_init loop_ctrl->ab.n nn_bind error \r\n");
		        return -1;
		    }            
        }
        break;
		case IPC_PLAT2MID: {
		    //创建2个通道 pipe和pub/sub
		    char addr[30] = {0};

		    loop_ctrl->ba.n = nn_socket(AF_SP, NN_PUSH);
            if (loop_ctrl->ba.n < 0) {
                ERROR_PRINT("loop_ctrl->ba.n =%d \r\n",loop_ctrl->ba.n);
                return -1;
            }
            
            loop_ctrl->ab.n = nn_socket(AF_SP, NN_SUB);
            if (loop_ctrl->ab.n < 0) {
                ERROR_PRINT("loop_ctrl->ab.n =%d \r\n",loop_ctrl->ab.n);
                return -1;
            }
            
		    sprintf(addr, GW2CCU_PIPE, ip);
            if (nn_connect(loop_ctrl->ba.n, addr) < 0) {//pipe
                ERROR_PRINT("nn_connect error ---2\r\n");
		        return -1;
		    }
		    //订阅
		    memset(addr,0, sizeof(addr));
		    memcpy(addr, loop_ctrl->subscrStr, strlen(loop_ctrl->subscrStr));
		    memcpy(addr + strlen(loop_ctrl->subscrStr), FILTERSTR, strlen(FILTERSTR));
		    if (nn_setsockopt(loop_ctrl->ab.n, NN_SUB, NN_SUB_SUBSCRIBE, addr, strlen(addr)) < 0) {
                ERROR_PRINT("nn_setsockopt failed ");
                return -1;
            }
            
            memset(addr,0, sizeof(addr));
		    sprintf(addr, GW2CCU_PUBSUB, ip);
		    if (nn_connect(loop_ctrl->ab.n, addr) < 0) {//sub
                ERROR_PRINT("nn_connect error ---3\r\n");
		        return -1;
		    }
		    
        }
        break;
		
		case IPC_MID2PLAT: {
		    //创建2个通道 pipe和pub/sub
		    char addr[30] = {0};

		    loop_ctrl->ba.n = nn_socket(AF_SP, NN_PULL);
            if (loop_ctrl->ba.n < 0) {
                ERROR_PRINT("__nanomsg_init loop_ctrl->ba.n =%d \r\n",loop_ctrl->ba.n);
                return -1;
            }
            
            loop_ctrl->ab.n = nn_socket(AF_SP, NN_PUB);
            if (loop_ctrl->ab.n < 0) {
                ERROR_PRINT("__nanomsg_init loop_ctrl->ab.n =%d \r\n",loop_ctrl->ab.n);
                return -1;
            }
            
		    sprintf(addr, GW2CCU_PIPE, ip);
            if (nn_bind(loop_ctrl->ba.n, addr) < 0) {//pipe
		        return -1;
		    }

		    memset(addr,0, sizeof(addr));
		    sprintf(addr, GW2CCU_PUBSUB, ip);
		    if (nn_bind(loop_ctrl->ab.n, addr) < 0) {//pub
		        return -1;
		    }
        }
        break;
        default: {
        }
        break;
    } 

	size_t size = sizeof(size_t);
	if (IPC_PLAT2MID == type || IPC_APP2MID == type){
		printf("[%s][%d]\n",__FUNCTION__,__LINE__);
		if (nn_getsockopt(loop_ctrl->ab.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&loop_ctrl->ab.s, &size) < 0) {
			ERROR_PRINT("nn_getsockopt IPC_PLAT2MID loop_ctrl->ab.s = %d \n", loop_ctrl->ab.s);
			return -1;
		}
	}else{
        if (nn_getsockopt(loop_ctrl->ba.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&loop_ctrl->ba.s, &size) < 0) {
            ERROR_PRINT("nn_getsockopt IPC_PLAT2MID loop_ctrl->ba.s = %d \n", loop_ctrl->ba.s);
            return -1;
        }
    }

	return 0;
}


/*=================================
*  TODO: only one methd pair
*
*
==================================*/
void loop_thread(void *arg){
    INFO_PRINT("loop_thread start!\r\n");
    ev_run (gloop, 0);
	
}

/*=================================
*  for gateway, the "id" and "ip" is necessary
*     chlMark: is unique, and suggest use gateway deviceCode
*     ip: ccu ip
*  for the ccu, id and ip are null
*
*
==================================*/
int kk_ipc_init(ipc_type type, ipc_cb cb, char* chlMark, char* ip)
{

	Bloop_ctrl_t* loop_ctrl;
    if (IPC_MID2PLAT == type){
         loop_ctrl = &Mloop_ctrl;
	}else {
		loop_ctrl = &Bloop_ctrl;
	}

    if (type == IPC_PLAT2MID && (chlMark == NULL || ip == NULL)){
        ERROR_PRINT("parameter is error \r\n");
        return -1;
    }

	if(loop_ctrl->cb != NULL){
		WARNING_PRINT("middleware to platform ipc has been inited!\r\n");
		return -1;
	}

	if (__nanomsg_init(loop_ctrl, type, chlMark, ip) < 0) {
		ERROR_PRINT("nanomsg init failed\r\n");
		return -1;
	}


    if (gloop == NULL){
        gloop = ev_loop_new(EVBACKEND_EPOLL);
        if (NULL == gloop) {
            ERROR_PRINT("create loop failed\r\n");
            return -1;
        }
        
    }

    loop_ctrl->type =  type;
    __loop_init(loop_ctrl, gloop);

	if (g_pTh == 0 && 0 != pthread_create(&g_pTh, NULL, loop_thread, NULL)) {
        ERROR_PRINT("create pthread failed\r\n");
        return -1;
    }

	loop_ctrl->cb = cb;
	loop_ctrl->type =  type;
    return 0;
	
}

int kk_ipc_deinit(ipc_type type)
{
    Bloop_ctrl_t* loop_ctrl;
    if (Bloop_ctrl.type == type){
        loop_ctrl = &Bloop_ctrl;
    }else if (Mloop_ctrl.type == type){
        loop_ctrl = &Mloop_ctrl;
    }else{
        ERROR_PRINT("kk_ipc_dinit failed, no ipc need destroy!\r\n");
        return -1;
    }
    
    if (loop_ctrl->ba.n > -1){
		nn_shutdown(loop_ctrl->ba.n, 0);
	}
	if (loop_ctrl->ab.n > -1){
		nn_shutdown(loop_ctrl->ab.n, 0);
	}

	ev_io_stop(gloop, &loop_ctrl->watcher);
    loop_ctrl->cb = NULL;
    loop_ctrl->type = IPC_UNDEF;
    
    if (Bloop_ctrl.cb ==NULL && Mloop_ctrl.cb ==NULL){
        ev_break(gloop,EVBREAK_ALL);
        pthread_exit(0);
        ev_loop_destroy(gloop);
        gloop = NULL;
        g_pTh = 0;
    }
    
	return 0;

}

int kk_ipc_send(ipc_type type, void* data, int len)
{
	char *chalMark = NULL;
	if (type == IPC_MID2PLAT){
		ERROR_PRINT(" type=IPC_MID2PLAT, please use kk_ipc_send_ex() api");
		return -1;
	}else if(type == IPC_PLAT2MID){
        if (strlen(Bloop_ctrl.subscrStr) > 0){
            chalMark = Bloop_ctrl.subscrStr;
        }
	}

	return kk_ipc_send_ex(type, data, len, chalMark);

}


int kk_ipc_send_ex(ipc_type type, void* data, int len, char* chalMark)
{
	int ret = 0;
	if (data != NULL){
		int filterlen = 0;
		void* buf = NULL;
		Bloop_ctrl_t* loop_ctrl;

		if (chalMark != NULL){
			filterlen = strlen(chalMark) + strlen(FILTERSTR);
		}

		buf = nn_allocmsg(len+filterlen, 0);
		if (buf == NULL){
			ERROR_PRINT("nn_allocmsg failed");
			return -1;
		}
		if (filterlen > 0){
			memcpy(buf, chalMark, strlen(chalMark));
			memcpy(buf + strlen(chalMark), FILTERSTR, strlen(FILTERSTR));
		}
		
		memcpy(buf + filterlen, data, len);
		if (type == IPC_MID2PLAT ){
			loop_ctrl = &Mloop_ctrl;
			ret =nn_send(Mloop_ctrl.ab.n, &buf, NN_MSG, NN_DONTWAIT);//NN_DONTWAIT
		}else if(type == IPC_MID2APP){
			loop_ctrl = &Bloop_ctrl;
			ret = nn_send(Bloop_ctrl.ab.n, &buf, NN_MSG, NN_DONTWAIT);
		}
		else{
			loop_ctrl = &Bloop_ctrl;
			ret = nn_send(Bloop_ctrl.ba.n, &buf, NN_MSG, NN_DONTWAIT);
		}

		if (ret < 0){
			ERROR_PRINT("nn_send failed with error code %d, str=%s \n", nn_errno(), nn_strerror(nn_errno()));
			loop_ctrl->sendErrCnt++;
			loop_ctrl->isconnect = 0;
			nn_freemsg(buf);
		}else{
			loop_ctrl->sendErrCnt = 0;
			loop_ctrl->isconnect = 1;
		}
	}

	return 0;
}


int kk_ipc_get_senderrcnt(ipc_type type){
    Bloop_ctrl_t* loop_ctrl;
    
    if (IPC_MID2PLAT == type){
         loop_ctrl = &Mloop_ctrl;
    }else {
         loop_ctrl = &Bloop_ctrl;
    }

    return loop_ctrl->sendErrCnt;
}
int kk_ipc_isconnect(ipc_type type){
    Bloop_ctrl_t* loop_ctrl;
    
    if (IPC_MID2PLAT == type){
         loop_ctrl = &Mloop_ctrl;
    }else {
         loop_ctrl = &Bloop_ctrl;
    }

    loop_ctrl->isconnect = 0;
    {
        
        for(int i =0; i<20;i++){
            kk_ipc_send(type, MAGIC, strlen(MAGIC)+1);
            usleep(500000);
            if (loop_ctrl->isconnect == 1){
                break;
            }
        }
        return loop_ctrl->isconnect;
    }
}

