#include <stdio.h>
#include <pthread.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <stdint.h>
#include <fcntl.h>
#include <errno.h>
#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <string.h>
#include "com_api.h"
#include "kk_log.h"



typedef struct{
    void *mutex;
	char deviceCode[DEVICE_CODE_LEN];
	char ip[MAX_IP_LEN];
	int sock;
	int isConnect;
}kk_tcp_ctrl_t;

static kk_tcp_ctrl_t g_tcp_ctrl[MAX_LISTEN_NUM];
static struct ev_loop *g_loop = NULL;
static pthread_t g_pTh = 0;
static int g_init = 0;
static ipc_cb* g_cb = NULL;
static struct ev_io w_accept;

static void *_MutexCreate(void)
{
    int err_num;
    pthread_mutex_t *mutex = (pthread_mutex_t *)malloc(sizeof(pthread_mutex_t));
    if (NULL == mutex) {
        return NULL;
    }

    if (0 != (err_num = pthread_mutex_init(mutex, NULL))) {
        INFO_PRINT("create mutex failed");
        free(mutex);
        return NULL;
    }

    return mutex;
}

static void _MutexDestroy(void *mutex)
{
    int err_num;

    if (!mutex) {
        INFO_PRINT("mutex want to destroy is NULL!");
        return;
    }
    if (0 != (err_num = pthread_mutex_destroy((pthread_mutex_t *)mutex))) {
        INFO_PRINT("destroy mutex failed");
    }

    free(mutex);
}

static void _MutexLock(void *mutex)
{
    int err_num;
    if (0 != (err_num = pthread_mutex_lock((pthread_mutex_t *)mutex))) {
        INFO_PRINT("lock mutex failed: - '%s' (%d)", strerror(err_num), err_num);
    }
}

static void _MutexUnlock(void *mutex)
{
    int err_num;
    if (0 != (err_num = pthread_mutex_unlock((pthread_mutex_t *)mutex))) {
        INFO_PRINT("unlock mutex failed - '%s' (%d)", strerror(err_num), err_num);
    }
}

static char *_next_json(char *str, int* hasFloor)
{
    char *ptr = str;
    int floor = 0;
    // judge if inside the "..."
    int quotes = 0;    
    if(str == NULL) {
        return NULL;
    }

    while(*ptr) {
        if(*ptr++ == '{') {
            ++floor;
            break;
        }
    }
    
    *hasFloor = floor;
    if(!floor) {
        //print_json();
        return NULL;
    }
    while(floor && *ptr) {
        switch(*ptr++) {
            case '{':{
                if(quotes)
                    break;
                ++floor;
                //printf("floor is %d \n", floor);
                break;
            }
            case '}':{
                if(quotes)
                    break;
                --floor;
                //printf("floor is %d \n", floor);            
                break;
            }
            case '"':{
                quotes = !quotes;
            }

            /* notice: judgment of '\\' added new without test*/
            case '\\':{ 
                break;
                ++ptr;
                break;
            }

            default:{
                break;
            }
        }
    }
    if(floor == 0) {
        return ptr;
    }
    return NULL;
}


static char* __do_data(char* buf, int buflen,ipc_cb* cb, char* deviceCode){
    char* endIdx = NULL;
    char* startIdx = buf;
    int hasfloor = 0;
    char buf2[1024] = {0};
    int cplen = 0;
    while(startIdx < buf + buflen){
        hasfloor = 0;
        endIdx = _next_json(startIdx,&hasfloor);
        if(endIdx == NULL && hasfloor == 0){//一般数据
            if (cb != NULL){
                cb(startIdx, strlen(startIdx),deviceCode);
            }
            return NULL;
            //break;
        }else if (endIdx == NULL){ //有json数据没接收完
            return startIdx;
        }
        
        cplen = endIdx - startIdx;
        memset(buf2,0, sizeof(buf2));
        memcpy(buf2,startIdx, cplen);

        if (cb != NULL){
            cb(buf2,cplen,deviceCode);
        }
        startIdx = endIdx;
    }
    return NULL;
}

static char* __do_spilt(char* buf, int buflen,ipc_cb* cb, char* deviceCode){

    char* pInx = buf;
	int slen = -1;

	while (buflen > 0){
        pInx += slen + 1;
        slen = strlen(pInx);
        INFO_PRINT("read_cb slen ====================================slen[%d]  \n",slen);
        if (cb != NULL){
			cb(pInx,slen,deviceCode);
		}
		buflen -= slen + 1;
	}

	if (buflen != 0 ){
        return pInx;
	}

	return NULL;
}

#if 0
// Save/Load the gateway list
static void kk_gw_list_save(void)
{
	FILE *fp;
	int len = MAX_LISTEN_NUM * sizeof(kk_tcp_ctrl_t);

	// Save device table
	fp = fopen("gwlist.txt", "w");
    fwrite(g_tcp_ctrl,len,1,fp);
	fclose(fp);

}


static void kk_gw_list_load(void)
{
	uint16_t i;
	FILE *fp;
    int readLen = MAX_LISTEN_NUM * sizeof(kk_tcp_ctrl_t);
    int retLen = 0;
    //char buf[1024] = {0};
	fp = fopen("gwlist.txt", "r");
	if (!fp) {
	    INFO_PRINT("open gatewaylist.txt failed! \n");
		goto error1;
	}

    retLen = fread(g_tcp_ctrl, readLen, 1, fp);
    INFO_PRINT("read gatewaylist.txt retLen= %d, readLen=%d ! \n", retLen,readLen);
	if (retLen !=  readLen ){
        INFO_PRINT("read gatewaylist.txt failed! \n");
	}

	fclose(fp);

	// Set the rest of the device table to null.
error1:
	for (i=0; i < MAX_LISTEN_NUM; i++) {
	    INFO_PRINT("deviceCode ip sock [%s] [%s] [%d] \n",g_tcp_ctrl[i].deviceCode,
	        g_tcp_ctrl[i].ip==NULL?"":g_tcp_ctrl[i].ip, g_tcp_ctrl[i].sock);
		g_tcp_ctrl[i].sock = -1;
	}

}


static int get_idx_by_ip(char ip[MAX_IP_LEN]){
    int i = 0;

    if (ip == NULL || strlen(ip) == 0){
        return -1;
    }
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(ip, g_tcp_ctrl[i].ip) == 0){
            return i;
       } 
    }
    return -1;
    
} 
#endif
static kk_tcp_ctrl_t* get_channel_by_deviceCode(char deviceCode[DEVICE_CODE_LEN]){
    int i = 0;

    if (deviceCode == NULL || strlen(deviceCode) == 0){
        return NULL;
    }
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(deviceCode, g_tcp_ctrl[i].deviceCode) == 0){
            return  &g_tcp_ctrl[i];
       } 
    }
    return NULL;
    
} 

static kk_tcp_ctrl_t* get_channel_by_ip(char ip[MAX_IP_LEN]){
    int i = 0;

    if (ip == NULL || strlen(ip) == 0){
        return NULL;
    }

    INFO_PRINT("[%s] ip=%s \n", __FUNCTION__,ip);
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(ip, g_tcp_ctrl[i].ip) == 0){
            INFO_PRINT("[%s] idx=%d ip=%s sock=%d\n", __FUNCTION__,i, ip, g_tcp_ctrl[i].sock);
            return  &g_tcp_ctrl[i];
       } 
    }
    return NULL;
    
} 

#if 0
static int set_status_by_ip(char ip[MAX_IP_LEN], int status){
    int i = 0;

    if (ip == NULL || strlen(ip) == 0){
        return -1;
    }
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(ip, g_tcp_ctrl[i].ip) == 0){
            g_tcp_ctrl[i].isConnect = status;
            return 0;
       } 
    }
    return -1;
    
}
#endif
static int reset_by_sock(int sock){
    int i = 0;

    if (sock < 0){
        return -1;
    }
    for(;i < MAX_LISTEN_NUM; i++){
       if(sock == g_tcp_ctrl[i].sock){
            g_tcp_ctrl[i].isConnect = 0;
			g_tcp_ctrl[i].sock = -1;
			_MutexDestroy(g_tcp_ctrl[i].mutex);
			g_tcp_ctrl[i].mutex = NULL;
            return 0;
       } 
    }
    return -1;
    
}
static int set_sock_by_ip(char ip[MAX_IP_LEN], int sock){
    int i = 0;

    if (ip == NULL || strlen(ip) == 0){
        return -1;
    }
    INFO_PRINT("[%s] ip=%s \n", __FUNCTION__,ip);
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(ip, g_tcp_ctrl[i].ip) == 0){
            
            g_tcp_ctrl[i].sock = sock;
            INFO_PRINT("[%s] idx=%d ip=%s sock=%d\n", __FUNCTION__,i, ip, g_tcp_ctrl[i].sock);
            //create mutex
            if (g_tcp_ctrl[i].mutex == NULL){
                g_tcp_ctrl[i].mutex = _MutexCreate();
                if (g_tcp_ctrl[i].mutex == NULL) {
                    INFO_PRINT("[%s] _MutexCreate failed \n", __FUNCTION__);
                    return -1;
                }
            }
            
            return 0;
       } 
    }
    return -1;
    
}

int kk_is_tcp_channel(char devCode[DEVICE_CODE_LEN]){
    
    int i = 0;

    if (devCode == NULL || strlen(devCode) == 0){
        return -1;
    }
    INFO_PRINT("[%s] devCode=%s \n", __FUNCTION__,devCode);
    for(;i < MAX_LISTEN_NUM; i++){
       if(strcmp(devCode, g_tcp_ctrl[i].deviceCode) == 0){
            
            INFO_PRINT("[%s] idx=%d ip=%s sock=%d\n", __FUNCTION__,i, g_tcp_ctrl[i].ip, g_tcp_ctrl[i].sock);
            return g_tcp_ctrl[i].sock;
       } 
    }
    return -1;
}

int kk_set_tcp_channel_by_idx(int idx, char devCode[DEVICE_CODE_LEN], char ip[MAX_IP_LEN]){
    if(idx >= MAX_LISTEN_NUM){
        INFO_PRINT("kk_set_tcp_channel_by_idx idx[%d] need less than %d \n", idx, MAX_LISTEN_NUM);
        return - 1;
    }
    
    if (devCode == NULL || strlen(devCode) == 0 || ip == NULL || strlen(ip) == 0){
        return -1;
    }
    //printf("[%s] devCode=%s \n", __FUNCTION__,devCode);
    memcpy(g_tcp_ctrl[idx].deviceCode, devCode, strlen(devCode));
    memcpy(g_tcp_ctrl[idx].ip, ip, strlen(ip));
    
    return 0;
}


int kk_set_tcp_channel(char devCode[DEVICE_CODE_LEN], char ip[MAX_IP_LEN]){
    int i = 0;
	int isEmptyIdx = -1;

    if (devCode == NULL || strlen(devCode) == 0 || ip == NULL || strlen(ip) == 0){
        INFO_PRINT("paramenter error \n");
        return -1;
    }
	
    for(;i < MAX_LISTEN_NUM; i++){

       if(strcmp(devCode, g_tcp_ctrl[i].deviceCode) == 0){
            strncpy(g_tcp_ctrl[i].ip, ip, strlen(ip));
            INFO_PRINT("find and replace it [%d][%s][%s] \n",i, g_tcp_ctrl[i].ip, devCode);
            //kk_gw_list_save();
            break;
       } 

	   if (strlen(g_tcp_ctrl[i].deviceCode) == 0 && isEmptyIdx == -1){
			isEmptyIdx = i;
		}
    }

	if (i < MAX_LISTEN_NUM){
		return 0;
	}

	if (isEmptyIdx != -1){
        
		strncpy(g_tcp_ctrl[isEmptyIdx].ip, ip, strlen(ip));
		strncpy(g_tcp_ctrl[isEmptyIdx].deviceCode, devCode, strlen(devCode));
		INFO_PRINT("idx deviceCode ip[%d][%s][%s]",isEmptyIdx, g_tcp_ctrl[isEmptyIdx].deviceCode, g_tcp_ctrl[isEmptyIdx].ip);
		//kk_gw_list_save();
		return 0;
	}
	
    return -1;
    
}




/*初始化服务端*/
static int server_socket_init(int *sd, char *ipaddr, uint16_t port)
{
    //创建socket
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (-1 == sock){
        INFO_PRINT("error socket \n");
        goto err1;
    }
    //设置立即释放端口并可以再次使用
    int reuse = 1;
    if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))){
        INFO_PRINT("error setsockopt \n");
        goto err2;
    }
    //设置为非阻塞
    if (-1 == fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK)){
        INFO_PRINT("================== fcntl \n");
        goto err2;
    }
    struct sockaddr_in addr;
    memset(&addr, 0 , sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    if (NULL == ipaddr) {
        addr.sin_addr.s_addr = htonl(INADDR_ANY);
    } else {
        addr.sin_addr.s_addr = inet_addr(ipaddr);
    }
    //绑定监听
    if (-1 == bind(sock, (struct sockaddr *)&addr, sizeof(addr))){
        INFO_PRINT("error bind \n");
        goto err2;
    }
    if (-1 == listen(sock, MAX_LISTEN_NUM)){
        INFO_PRINT("error listen \n");
        goto err2;
    }
    *sd = sock;
    return 0;
err2:
    close(sock);
err1:
    return -1;
}
 
/*读回调*/
static void read_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
{
    INFO_PRINT("================== read_cb [%d]\n", revents);
    char buffer[BUFFER_SIZE] = {0};
    if (EV_ERROR & revents) {
      INFO_PRINT("read got invalid event...\r\n");
      return;
    }
    int res = 0;
    int offset = 0;
	kk_tcp_ctrl_t* tcp_ctrl = (kk_tcp_ctrl_t*)watcher->data;
again:	
	_MutexLock(tcp_ctrl->mutex);
    int32_t bytes = recv(watcher->fd, buffer + offset, sizeof(buffer) - offset - 1,0);
    _MutexUnlock(tcp_ctrl->mutex);
    if (-1 == bytes) {
        //tcp Error
        if (EINTR != errno && EAGAIN != errno) {
            res = 1;
        }
    } else if (0 == bytes) {
        //tcp Close
        res = 2;
    }
    if (0 != res) {
        //关闭事件循环并释放watcher
        INFO_PRINT("TCP CLOSE\r\n");
        reset_by_sock(watcher->fd);
        ev_io_stop(loop,watcher);
        free(watcher);
    } else {
        //printf("READ:\r\n %s =====[%d]\r\n", buffer, bytes);
        //printf("read_cb deviceCode ip sock [%s][%s][%d] \n",tcp_ctrl->deviceCode,tcp_ctrl->ip, tcp_ctrl->sock);
		/*if (g_cb != NULL){
			g_cb(buffer,bytes,tcp_ctrl->deviceCode);
		}*/
		char* retpst = NULL;
		if (strlen(buffer) == bytes){
    		retpst = __do_data(buffer, bytes + offset, g_cb, tcp_ctrl->deviceCode);
            if (retpst != NULL){
                offset = bytes + offset - (retpst - buffer);
                memmove(buffer, retpst, offset);
                memset(buffer + offset, 0, sizeof(buffer) - offset);
                INFO_PRINT("====read not complete, need again offset=%d \n",offset);
                goto again;
            }
        }
        else{
         //处理‘\0’ 结束符粘包
            retpst = __do_spilt(buffer, bytes + offset, g_cb, tcp_ctrl->deviceCode);
            if (retpst != NULL){
                offset = bytes + offset - (retpst - buffer);
                memmove(buffer, retpst, offset);
                memset(buffer + offset, 0, sizeof(buffer) - offset);
                INFO_PRINT("====read not complete, need again offset=%d \n",offset);
                goto again;
            }
            
    		/*char* pInx = buffer;
    		int buflen = offset + bytes;
    		int slen = -1;

    		while (buflen > 0){
                pInx += slen + 1;
                slen = strlen(pInx);
                printf("read_cb slen ====================================slen[%d]  \n",slen);
                if (g_cb != NULL){
        			g_cb(pInx,slen,tcp_ctrl->deviceCode);
        		}
        		buflen -= slen + 1;
    		}

    		if (buflen != 0 && bytes == (BUFFER_SIZE - offset)){
                printf("read_cb need recv agian====================================[%d]  \n",buflen);
                offset = buflen + slen + 1;
                memmove(buffer,pInx,offset);
                memset(buffer + offset, 0, (BUFFER_SIZE - offset));
                goto again;
    		}*/
        }
		
    }
}
 
 
/*accept回调函数*/
static void accept_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
{
    INFO_PRINT("================== accept_cb \n");
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    if (EV_ERROR & revents) {
      INFO_PRINT("accept got invalid event...\r\n");
      return;
    }
    //accept连接
    int sock = accept(watcher->fd, (struct sockaddr *)&client_addr, &client_len);
    if (-1 == sock) {
        return;
    }
    //设置非阻塞
    if(-1 == fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK)) {
        close(sock);
        return;
    }

    if(set_sock_by_ip(inet_ntoa(client_addr.sin_addr),sock) < 0){
        //ip无效，reject it
        INFO_PRINT("This ip[%s][%u] is invaild, reject it! \r\n", inet_ntoa(client_addr.sin_addr), client_addr.sin_port);
        close(sock);
        return;
    }
    INFO_PRINT("Successfully connected with client: %s:%u\r\n", \
    inet_ntoa(client_addr.sin_addr), client_addr.sin_port);
    
    //加入事件循环
    struct ev_io *w_client = (struct ev_io*) malloc (sizeof(struct ev_io));
    if (w_client == NULL){
        INFO_PRINT("malloc w_client failed \r\n");
        return;
    }
    
	w_client->data = get_channel_by_ip(inet_ntoa(client_addr.sin_addr));
    ev_io_init(w_client, read_cb, sock, EV_READ);
    ev_io_start(loop, w_client);
}




/*int run()
{
    int sd;
    struct ev_io w_accept;
    g_loop = ev_loop_new(EVBACKEND_EPOLL);
    if (NULL == g_loop) {
        printf("loop create failed\r\n");
        return -1;
    }
    if (server_socket_init(&sd, NULL, CCU_TCP_PORT) < 0) {
        printf("server init failed\r\n");
        return -1;
    }
    ev_io_init(&w_accept, accept_cb, sd, EV_READ);
    ev_io_start(g_loop, &w_accept);
    ev_run(g_loop, 0);
    return 0;
}*/

static void loop_tcp_thread(void *arg){
    INFO_PRINT("loop_tcp_thread start!\r\n");    
    int sd;
    
    g_loop = ev_loop_new(EVBACKEND_EPOLL);
    if (NULL == g_loop) {
        INFO_PRINT("loop create failed\r\n");
        return;
    }
    if (server_socket_init(&sd, NULL, CCU_TCP_PORT) < 0) {
        INFO_PRINT("server init failed\r\n");
        return;
    }
    ev_io_init(&w_accept, accept_cb, sd, EV_READ);
    ev_io_start(g_loop, &w_accept);
    ev_run (g_loop, 0);
    close(sd);
    INFO_PRINT("loop_tcp_thread================== end \n");
}


int kk_tcp_channel_ser_send(char* data, int len, char chalMark[DEVICE_CODE_LEN]){
    int ret = 0;
    fd_set fds;
    struct timeval timeout={0,200}; //select等待3秒，3秒轮询，要非阻塞就置0 
    
    if (data != NULL){
        kk_tcp_ctrl_t* chl_ctrl = get_channel_by_deviceCode(chalMark);
        if (NULL == chl_ctrl){
            INFO_PRINT("[%s] get_channel_by_deviceCode is NULL!!! \n",__FUNCTION__);
            return -1;
        }
        
        if (chl_ctrl->sock > -1){
            FD_ZERO(&fds); //每次循环都要清空集合，否则不能检测描述符变化
            FD_SET(chl_ctrl->sock,&fds); //添加描述符 
            switch(select(chl_ctrl->sock + 1,NULL,&fds,NULL,&timeout)) //select使用 
            { 
                case -1: 
                     chl_ctrl->isConnect = 0;
                     INFO_PRINT(" [%s] select error ret=%d \n", __FUNCTION__, -1);
                     break; //select错误 退出循环 
                case 0:
                    INFO_PRINT(" [%s] select skip=================== ret=%d \n", __FUNCTION__, 0);
                    break; //再次轮询 
                
                default: 
                if(FD_ISSET(chl_ctrl->sock,&fds)) //测试sock是否可读，即是否网络上有数据 
                { 
                    _MutexLock(chl_ctrl->mutex);
                    ret = write(chl_ctrl->sock, data, len);
                    _MutexUnlock(chl_ctrl->mutex);
                    
                    if( ret <= 0){
                        INFO_PRINT("=================write error ret=%d \n",ret);
                        if (errno != EINTR){
                             chl_ctrl->isConnect = 0;
                             INFO_PRINT("write error reconnect!! \n");
                             break;
                        }
                    }
                }
                break;

            }// end switch 
            /*_MutexLock(chl_ctrl->mutex);
            printf("[%s] kk_tcp_channel_ser_send [%s] \n",__FUNCTION__,data);
            ret = write(chl_ctrl->sock, data, len);
            _MutexUnlock(chl_ctrl->mutex);
            if (ret < 0){
                printf("[%s] write failed!!!! \n",__FUNCTION__);
            }*/
        }
        
    }
    return ret;
}


int kk_TCP_channel_init(ipc_cb cb)
{
    int i = 0;
	int sd;    

	if (g_init == 1){
		INFO_PRINT("kk_TCP_channel_init has been inited \n");
		return -1;
	}
	g_init = 1;
	//memset(g_tcp_ctrl, 0, sizeof(kk_tcp_ctrl_t)*MAX_LISTEN_NUM);
	//kk_gw_list_load();

    for(i = 0; i < MAX_LISTEN_NUM; i++){
        g_tcp_ctrl[i].sock = -1;
	}

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

    return 0;


}


int kk_TCP_channel_deinit(ipc_type type)
{
    int i = 0;
	
	for(i = 0; i < MAX_LISTEN_NUM; i++){
		if(g_tcp_ctrl[i].sock > -1){
			//close(g_tcp_ctrl[i].sock);
		}
		if(g_tcp_ctrl[i].mutex != NULL){
			_MutexDestroy(g_tcp_ctrl[i].mutex);
			g_tcp_ctrl[i].mutex = NULL;
		}
	}

	ev_io_stop(g_loop, &w_accept);
	ev_break(g_loop,EVBREAK_ALL);
    pthread_exit(0);
    ev_loop_destroy(g_loop);
    g_loop = NULL;
    g_pTh = 0;
    g_init = 0;
    

    
	return 0;

}


//client
/*初始化服务端*/
typedef struct {
    void *mutex;
    int   sd;
    int isConnect;
    char ip[MAX_IP_LEN];
    int port;
    ipc_cb* cb;
    int retry;
} kk_tcp_client_t;

static kk_tcp_client_t g_client_ctrl = {NULL, -1, 0,{0},0, NULL,0};

static int _init_client(){
    memset(&g_client_ctrl, 0 ,sizeof(kk_tcp_client_t));
    /* Create Mutex */
    g_client_ctrl.mutex = _MutexCreate();
    if (g_client_ctrl.mutex == NULL) {
        return -1;
    }
    g_client_ctrl.sd = -1;
    return 0;
}


static int client_socket_init(int *sd, char *ipaddr, uint16_t port)
{
    //创建socket
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (-1 == sock){
        INFO_PRINT("error socket \n");
        goto err1;
    }
    //设置立即释放端口并可以再次使用
    int reuse = 1;
    if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))){
        INFO_PRINT("error setsockopt \n");
        goto err2;
    }
    //设置为非阻塞
    if (-1 == fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK)){
        INFO_PRINT("================== fcntl \n");
        goto err2;
    }
    struct sockaddr_in addr;
    memset(&addr, 0 , sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    if (NULL == ipaddr) {
        addr.sin_addr.s_addr = htonl(INADDR_ANY);
    } else {
        addr.sin_addr.s_addr = inet_addr(ipaddr);
    }
    //连接
    int retry = 0;
    for(;retry < 3; retry++){
        if(-1 != connect(sock, (struct sockaddr *)&addr, sizeof(addr))){
            break;
        }
        //printf("==================connect retry=%d \n", retry);
        sleep(1);
    }
    if (retry >= 3){
        INFO_PRINT("==================connect failed \n");
        goto err2;
    }
    
    INFO_PRINT("==================[%s] ok sock=%d \n", __FUNCTION__, sock);
    *sd = sock;
    return 0;
err2:
    close(sock);
err1:
    return -1;
}

static void loop_tcp_client_thread(void *arg){
    INFO_PRINT("loop_tcp_client_thread start!\r\n"); 
    char buf[1024]= {0};
    int ret = 0;
    int offset = 0;
    fd_set fds;
    struct timeval timeout={0,200}; //select等待3秒，3秒轮询，要非阻塞就置0 
    
    while(1){
        if(-1 == client_socket_init(&g_client_ctrl.sd,g_client_ctrl.ip, g_client_ctrl.port)){
            INFO_PRINT("connect failed \n");
            sleep(1);
            g_client_ctrl.retry++;
            continue;
        }
        g_client_ctrl.isConnect = 1;
        g_client_ctrl.retry = 0;
        while(g_client_ctrl.isConnect){
            FD_ZERO(&fds); //每次循环都要清空集合，否则不能检测描述符变化
            FD_SET(g_client_ctrl.sd,&fds); //添加描述符 
            offset = 0;
            switch(select(g_client_ctrl.sd + 1,&fds,NULL,NULL,&timeout)) //select使用 
            { 
                case -1: 
                     g_client_ctrl.isConnect = 0;
                     INFO_PRINT(" [%s] select error ret=%d \n", __FUNCTION__, ret);
                     break; //select错误 退出循环 
                case 0:
                    break; //再次轮询 
                
                default: 
                if(FD_ISSET(g_client_ctrl.sd,&fds)) //测试sock是否可读，即是否网络上有数据 
                { 
                    //接受网络数据 
                    memset(buf,0x0,sizeof(buf));
                   clientRead: 
                    _MutexLock(g_client_ctrl.mutex);
					
                    ret = read(g_client_ctrl.sd, buf + offset, sizeof(buf)- offset - 1);
                    _MutexUnlock(g_client_ctrl.mutex);
                    
                    if( ret <= 0){
                        INFO_PRINT("=================read error ret=%d \n",ret);
                        if (errno != EINTR){
                             g_client_ctrl.isConnect = 0;
                             INFO_PRINT("read error reconnect!! \n");
                             break;
                        }
                    }else if(ret > 0){
                        //printf("11buf = %s\n",buf);
                        #if 0
                        if (g_client_ctrl.cb != NULL){
                            g_client_ctrl.cb(buf,ret,"");
                        }
                        #else
                        
                        char* retpst = __do_data(buf, ret + offset, g_client_ctrl.cb, "");
                        if (retpst != NULL){
                            offset = ret + offset - (retpst - buf);
                            memmove(buf, retpst, offset);
                            memset(buf + offset, 0, sizeof(buf) - offset);
                            INFO_PRINT("====read not complete, need again offset=%d \n",offset);
                            goto clientRead;
                        }
                        #endif
                    }
                }
                break;

            }// end switch 
            //usleep(100000);

        }
        INFO_PRINT("network error, try connect again! \n");
        close(g_client_ctrl.sd);
    }    
    INFO_PRINT("loop_tcp_client_thread================== end \n");
	
}

int kk_get_retry_num(){
    return g_client_ctrl.retry;
}
int kk_reset_retry_num(){
    return g_client_ctrl.retry = 0;
}


int kk_tcp_client_send(char* data, int len){
    int ret = 0;
    int cnt = 0;
    fd_set fds;
    struct timeval timeout={0,20000}; 

    if ( data != NULL){
        while(g_client_ctrl.sd == -1 && cnt < 5){
            INFO_PRINT("[%s] tcp don't connect, sleep 1s !!!! \n",__FUNCTION__);
            sleep(1);
            cnt++;
        }
        if (g_client_ctrl.sd < 0){
            INFO_PRINT("[%s] The tcp socket created fialid !!!! \n",__FUNCTION__);
            return -1;
        }

        FD_ZERO(&fds); //每次循环都要清空集合，否则不能检测描述符变化
        FD_SET(g_client_ctrl.sd,&fds); //添加描述符 
        switch(select(g_client_ctrl.sd + 1,NULL,&fds,NULL,&timeout)) //select使用 
        { 
            case -1: 
                 g_client_ctrl.isConnect = 0;
                 INFO_PRINT(" [%s] select error ret=%d \n", __FUNCTION__, ret);
                 break; //select错误 退出循环 
            case 0:
                INFO_PRINT(" [%s] select error ret=%d \n", __FUNCTION__, 0);
                break; //再次轮询 
            
            default: 
            if(FD_ISSET(g_client_ctrl.sd,&fds)) //测试sock是否可读，即是否网络上有数据 
            { 
                //接受网络数据 
                _MutexLock(g_client_ctrl.mutex);
                //printf("kk_tcp_client_send [%s] =========!\n",data);
                ret = write(g_client_ctrl.sd, data, len);
                _MutexUnlock(g_client_ctrl.mutex);
                
                if( ret <= 0){
                    INFO_PRINT("=================send error ret=%d \n",ret);
                    if (errno != EINTR){
                         g_client_ctrl.isConnect = 0;
                         INFO_PRINT("send error reconnect!! \n");
                         break;
                    }
                }
            }
            break;

        }// end switch 
        
        /*_MutexLock(g_client_ctrl.mutex);
        printf("kk_tcp_client_send [%s] \n",data);
        ret = write(g_client_ctrl.sd, data, len);
         _MutexUnlock(g_client_ctrl.mutex);
        if (ret < 0){
            printf("[%s] write failed ret=%d, reconnect !!!! \n",__FUNCTION__, ret);
            g_client_ctrl.isConnect = 0;
            ret = -1;
        }*/
    }
    return ret;
}
int kk_tcp_client_init(char ip[MAX_IP_LEN], int port, ipc_cb cb)
{
    //kk_tcp_client_deinit();
    _init_client();
    g_client_ctrl.port = port;
    memcpy(g_client_ctrl.ip, ip, strlen(ip));
    if (g_pTh ==0 && 0 != pthread_create(&g_pTh, NULL, loop_tcp_client_thread, NULL)) {
        INFO_PRINT("create pthread failed\r\n");
        return -1;
    };
	g_client_ctrl.cb = cb; 
    return 0;
}

void kk_tcp_client_deinit(void){
    if (g_client_ctrl.sd > -1){
        close(g_client_ctrl.sd);
        g_client_ctrl.sd = -1;
    }
    _MutexDestroy(g_client_ctrl.mutex);
}




