mbtk/libmbtk_lib/net/mbtk_sock2.c

#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <sys/time.h>
#include <fcntl.h>
#include <netdb.h>
#include <pthread.h>
#include <sys/ioctl.h>

#ifdef LOG_TAG
#undef LOG_TAG
#endif
#define LOG_TAG "mbtk_sock"
#include <mbtk_log.h>

#include "mbtk_sock2.h"
#include "mbtk_sock_internal.h"
#include "mbtk_str.h"
//#include "mbtk_openssl.h"

#define DFL_CA_FILE   "/ca.crt"
#define DFL_CRT_FILE  "/client.crt"
#define DFL_KEY_FILE  "/client.key"

#define SA struct sockaddr

// Must define LOG_TAG in the first.
//#include "mbtk_log.h"
static int epoll_fd = -1;
static int pipe_fds[2];
static struct epoll_event epoll_events[20];
static pthread_t sock_thread_id;
static bool sock_thread_running = FALSE;
static mbtk_sock_s *mbtk_sock[MBTK_HANDLE_MAX_NUM] = {NULL};

static int sock_find_first_free(const mbtk_sock_inter_info_s *inter_info)
{
    if(inter_info == NULL) {
        LOGE("inter_info is NULL.");
        return -1;
    }

    int index = 0;
    //while((inter_info + index)->fd > 0) {
    while(inter_info[index].fd > 0) {
        index++;
    }

    if(index == MBTK_SOCK_MAX_NUM) {
        LOGE("sock_infos too more.");
        return -1;
    }

    return index;
}

static bool sock_info_check(int handle,mbtk_sock_inter_info_s *inter_info)
{
    if(inter_info == NULL || mbtk_sock[handle] == NULL) {
        LOGE("internal_info is NULL.");
        return FALSE;
    }

    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(inter_info->fd ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            return TRUE;
        }
        index++;
    }

    return FALSE;
}

#if 0
static bool sock_is_close(int sockfd)
{
    char buff[32];
    int recvBytes = recv(sockfd, buff, sizeof(buff), MSG_PEEK);

    int err = errno;
    //cout << "In close function, recv " << recvBytes << " bytes, err " << sockErr << endl;

    if(recvBytes > 0) //Get data
        return FALSE;

    if((recvBytes == -1) && (err == EWOULDBLOCK)) //No receive data
        return FALSE;

    return TRUE;
}
#endif

static int sock_info_find_by_fd(int handle,int fd)
{
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(fd == mbtk_sock[handle]->inter_infos[index].fd) {
            return index;
        }
        index++;
    }

    return -1;
}

static void* sock_thread_run(void *data)
{
    LOGD("socket thread is running...");
    if(data != NULL)
        LOGD("sock_thread_run has arg.");

    int nready;
    if (socketpair( AF_UNIX, SOCK_STREAM, 0, pipe_fds) < 0) {
        LOGE("socketpair() error.");
        return NULL;
    } else {
        struct epoll_event ev;
        ev.data.fd = pipe_fds[1];
        ev.events = EPOLLIN | EPOLLET;
        epoll_ctl(epoll_fd,EPOLL_CTL_ADD,pipe_fds[1],&ev);
    }

    while(sock_thread_running) {
        nready = epoll_wait(epoll_fd,epoll_events,20,-1);
        int i;
        for(i=0;i<nready;++i) {
            LOGV("fd[%d] event = %x",epoll_events[i].data.fd,epoll_events[i].events);
            if(pipe_fds[1] == epoll_events[i].data.fd) {
                LOGD("Get exist sig.");
                sock_thread_running = FALSE;
                break;
            }

            int handle = 0;
            while(handle < MBTK_HANDLE_MAX_NUM) {
                if(mbtk_sock[handle] != NULL) {
                    int index = sock_info_find_by_fd(handle,epoll_events[i].data.fd);
                    if(index >= 0 && mbtk_sock[handle]->init_info.sock_cb != NULL) {
                        mbtk_sock_inter_info_s *inter_info = &(mbtk_sock[handle]->inter_infos[index]);
                        mbtk_sock_info *info = &(mbtk_sock[handle]->infos[index]);

                        //if(sock_is_close(epoll_events[i].data.fd)) {
                        //    LOGE("Socket %d is closed.",epoll_events[i].data.fd);
                        //    break;
                        //}
                        mbtk_sock_cb_info_s sock_info;
                        sock_info.sock_fd = inter_info->fd;
                        sock_info.event = epoll_events[i].events;
                        sock_info.sock_type = info->type;
                        mbtk_sock[handle]->init_info.sock_cb(handle, &sock_info);

                        if(epoll_events[i].events & EPOLLERR){ // error[ UDP server can't use.]
                            LOGD("%d EPOLLERR.",epoll_events[i].data.fd);
                        }

                        if ((epoll_events[i].events & EPOLLIN)
                            && (epoll_events[i].events & EPOLLOUT)) {
                            LOGD("%d can read and write.",epoll_events[i].data.fd);
                            int error = -1;
                            socklen_t len = sizeof(int);
                            if(getsockopt(epoll_events[i].data.fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0){
                                LOGE("getsockopt fail.[%d]",errno);
                            }else{
                                LOGD("error = %d",error);
                            }
                        }

                        if (epoll_events[i].events & EPOLLOUT) { // Can write.
                            LOGD("%d can write.",epoll_events[i].data.fd);
                        }

                        if (epoll_events[i].events & EPOLLIN) { // Can read.
                            LOGD("%d can read.",epoll_events[i].data.fd);
                        }
                    }
                }

                handle++;
            }
        }
    }

    LOGD("socket thread exit.");
    return ((void*)0);
}

static int sock_thread_start()
{
    sock_thread_running = TRUE;
    if (0 != pthread_create(&sock_thread_id, NULL, sock_thread_run, NULL))
    {
        LOGE("error when create pthread,%d\n", errno);
        return -1;
    }

    return 0;
}

void net_state_callback_func(mbtk_net_change_type_t type, const void *data)
{
    if(type == MBTK_NET_CHANGE_ADDR && data != NULL) {
        int handle = 0;
        const mbtk_net_addr_change_info_t *addr_info = (const mbtk_net_addr_change_info_t *)data;
        while(handle < MBTK_HANDLE_MAX_NUM) {
            if(mbtk_sock[handle] != NULL) {
                if(mbtk_sock[handle]->init_info.net_cb != NULL) {
                    mbtk_net_cb_info_s net_info;
                    net_info.state = (addr_info->type == MBTK_NET_ADDR_CHANGE_TYPE_ADD) ? 1 : 0;
                    net_info.addr = addr_info->addr;
                    net_info.if_name = addr_info->if_name;
                    mbtk_sock[handle]->init_info.net_cb(handle, &net_info);
                }
            }

            handle++;
        }
    }
}

#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
static int mbtk_sock_openssl_init(int fd, bool ingnore_cert, mbtk_openssl_info_s *inter_info)
{
    if(NULL == inter_info)
    {
        LOGE("[%s] inter_info [NULL]", __func__);
        return -1;
    }

    mbtk_openssl_result_e mbtk_ssl_ret = MBTK_OPENSSL_RESULT_SUCCESS;
    mbtk_openssl_options_s opt = {0};
    
    mbtk_openssl_options_default(&opt);
    if(!ingnore_cert)
    {
        opt.load_cert = true;
        opt.ca_file = DFL_CA_FILE;
        opt.crt_file = DFL_CRT_FILE;
        opt.key_file = DFL_KEY_FILE;
        opt.safety_level = MBTK_OPENSSL_SAFETY_LEVEL_0;
    }
    
    mbtk_ssl_ret = mbtk_openssl_init(fd, &opt, inter_info);
    if(MBTK_OPENSSL_RESULT_SUCCESS != mbtk_ssl_ret)
    {
        LOGE("[%s] mbtk_openssl_init() fail", __func__);
        return -1;
    }

    return 0;
}
#endif

extern mbtk_sock_handle mbtk_sock_init(mbtk_init_info *info)
{
    mbtk_sock_handle handle = 0;
    while(handle < MBTK_HANDLE_MAX_NUM) {
        if(mbtk_sock[handle] == NULL)
            break;

        handle++;
    }

    if(handle == MBTK_HANDLE_MAX_NUM) {
        LOGE("Socket handle is full.");
        return -1;
    }

    mbtk_sock[handle] = (mbtk_sock_s*)malloc(sizeof(mbtk_sock_s));
    memset(mbtk_sock[handle],0x0,sizeof(mbtk_sock_s));
    if(info != NULL) {
        mbtk_sock[handle]->init_info.net_type = info->net_type;
        mbtk_sock[handle]->init_info.net_cb = info->net_cb;
        mbtk_sock[handle]->init_info.sock_cb = info->sock_cb;
        if(!str_empty(info->if_name)) {
            memcpy(mbtk_sock[handle]->init_info.if_name, info->if_name, strlen(info->if_name));
        }
    } else {
        mbtk_sock[handle]->init_info.net_type = MBTK_NET_LINUX;
        mbtk_sock[handle]->init_info.net_cb = NULL;
        mbtk_sock[handle]->init_info.sock_cb = NULL;
    }

    if(!sock_thread_running) {
        epoll_fd = epoll_create(256);
        if(sock_thread_start()) {
            LOGE("Start thread fail.");
            return -1;
        }
    }

    if(mbtk_net_monitor_reg(str_empty(info->if_name) ? NULL : info->if_name, net_state_callback_func)) {
        LOGE("mbtk_net_monitor_reg() fail.");
        return -1;
    }

    return handle;
}

extern mbtk_sock_session mbtk_sock_open(mbtk_sock_handle handle,mbtk_sock_info *info,
                unsigned int timeout,
                int *mbtk_errno)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    *mbtk_errno = MBTK_SOCK_ERROR;
    if(info == NULL) {
        LOGE("mbtk_sock_info not be NULL.");
        return -1;
    }

    int index_free = sock_find_first_free(mbtk_sock[handle]->inter_infos);
    if(index_free < 0) {
        LOGE("sock_find_first_free() fail.");
        return -1;
    }

    memcpy(&(mbtk_sock[handle]->infos[index_free]),info,sizeof(mbtk_sock_info));
    if(info->type == MBTK_SOCK_UDP) { // UDP
        if((mbtk_sock[handle]->inter_infos[index_free].fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0){
            LOGE("socket() fail.[%d]",errno);
            goto result_fail;
        }
    } else { // TCP
        if((mbtk_sock[handle]->inter_infos[index_free].fd = socket(AF_INET, SOCK_STREAM, 0)) < 0){
            LOGE("socket() fail.[%d]",errno);
            goto result_fail;
        }
    }
    // Set O_NONBLOCK
    int flags = fcntl(mbtk_sock[handle]->inter_infos[index_free].fd, F_GETFL, 0);
    if (flags < 0) {
        LOGE("Get flags error:%s\n", strerror(errno));
        goto result_fail_with_close;
    }
    flags |= O_NONBLOCK;
    if (fcntl(mbtk_sock[handle]->inter_infos[index_free].fd, F_SETFL, flags) < 0) {
        LOGE("Set flags error:%s\n", strerror(errno));
        goto result_fail_with_close;
    }

    // Connect
    LOGD("Start conn:%s:%d",info->address,info->port);
    if(strlen(info->address) > 0 && info->port > 0) {
        if(strlen(info->local_address) > 0 || info->local_port > 0) {
            // 指定本地IP和端口，不指定内核会自动指定（一般不指定）
            struct sockaddr_in loc_addr;
            memset(&loc_addr, 0, sizeof(struct sockaddr_in));
            loc_addr.sin_family = AF_INET;

            // 指定IP
            if(strlen(info->local_address) > 0) {
                if(inet_pton(AF_INET, info->local_address, &loc_addr.sin_addr) < 0) {
                    LOGE("inet_pton() error:%d", errno);
                    goto result_fail_with_close;
                }
            }

            if(info->local_port > 0) {
                loc_addr.sin_port = htons(info->local_port);
            }
            if(bind(mbtk_sock[handle]->inter_infos[index_free].fd, (struct sockaddr *)&loc_addr, sizeof(loc_addr)) < 0) {
                LOGE("bind() error:%d", errno);
                if(errno == EADDRINUSE) { // 地址已在使用
                    LOGE("EADDRINUSE : Local port already occupied.");
                }
                goto result_fail_with_close;
            } else {
                LOGD("Bind ip/port success.");
            }
        }

        struct sockaddr_in servaddr;
        bzero(&servaddr, sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(info->port);

        struct hostent *he = gethostbyname(info->address);
        if (he == NULL){
            LOGE("gethostbyname() fail.[%d]",errno);
            goto result_fail_with_close;
        } else {
            LOGD("Ip : %s",he->h_addr_list[0]);
        }
        memcpy(&servaddr.sin_addr, he->h_addr_list[0], sizeof(struct in_addr));

        if(connect(mbtk_sock[handle]->inter_infos[index_free].fd, (SA *) &servaddr, sizeof(servaddr)) < 0){
            if(EINPROGRESS != errno){
                LOGE("connect() fail.[%d]",errno);
                goto result_fail_with_close;
            }
        }

        fd_set rset, wset;
        FD_ZERO(&rset);
        FD_ZERO(&wset);
        FD_SET(mbtk_sock[handle]->inter_infos[index_free].fd, &rset);
        FD_SET(mbtk_sock[handle]->inter_infos[index_free].fd, &wset);
        struct timeval time_out;
        time_out.tv_sec = timeout/1000;
        time_out.tv_usec = timeout%1000*1000;
        int nready = select(mbtk_sock[handle]->inter_infos[index_free].fd + 1,
                        &rset, &wset, NULL, &time_out);
        LOGD("nready = %d",nready);
        if(nready == 0){// Timeout
            LOGE("Timeout.");
            printf("Timeout.\n");
            goto result_fail_with_close;
        }else{
            if (FD_ISSET(mbtk_sock[handle]->inter_infos[index_free].fd, &rset)
                && FD_ISSET(mbtk_sock[handle]->inter_infos[index_free].fd, &wset)) {
                int error = -1;
                socklen_t len = sizeof(int);
                LOGE("Can read and write.");
                if(getsockopt(mbtk_sock[handle]->inter_infos[index_free].fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0){
                    LOGE("getsockopt fail.[%d]",errno);
                    goto result_fail_with_close;
                }else{
                    LOGE("error = %d",error);
                    if(error != 0){ // Fail
                        goto result_fail_with_close;
                    }
                }
            }else if(FD_ISSET(mbtk_sock[handle]->inter_infos[index_free].fd, &wset)){
                LOGI("Can write.");
                printf("Can write.\n");
            }else{
                LOGE("Can read(Impossible).");
                goto result_fail_with_close;
            }
        }
    } else {
        LOGE("Can not conn.");
        goto result_fail_with_close;
    }

    if(mbtk_sock[handle]->init_info.sock_cb) {
        struct epoll_event ev;
        ev.data.fd = mbtk_sock[handle]->inter_infos[index_free].fd;
        ev.events = EPOLLIN | EPOLLET;
        epoll_ctl(epoll_fd,EPOLL_CTL_ADD,mbtk_sock[handle]->inter_infos[index_free].fd,&ev);
    }
#if 1
    if(info->ftp_ssl_support)
    {
        if(info->is_support_ssl){
            mbtk_sock[handle]->infos[index_free].is_support_ssl = 0;
            unsigned char mbtk_ftp_ssl_read_buf_s[256];
            int err_rw;
                memset(mbtk_ftp_ssl_read_buf_s,0,sizeof(mbtk_ftp_ssl_read_buf_s));
                mbtk_sock_read(handle,mbtk_sock[handle]->inter_infos[index_free].fd,
                        mbtk_ftp_ssl_read_buf_s,
                        sizeof(mbtk_ftp_ssl_read_buf_s),
                        60000,
                        &err_rw);
                printf("\nmbtk_sock_read:\n%s\n",mbtk_ftp_ssl_read_buf_s);

            char cmd_buff[50];
            int len=0;
            memset(cmd_buff,0,sizeof(cmd_buff));

            len = snprintf(cmd_buff, 50, "AUTH TLS\r\n");
            cmd_buff[len] = '\0';
            //printf("\n cmd_buff = %s\n", cmd_buff);

            mbtk_sock_write(handle,mbtk_sock[handle]->inter_infos[index_free].fd,
                                                cmd_buff,
                                                strlen(cmd_buff),
                                                60000,
                                                &err_rw);

            memset(mbtk_ftp_ssl_read_buf_s,0,sizeof(mbtk_ftp_ssl_read_buf_s));
                mbtk_sock_read(handle,mbtk_sock[handle]->inter_infos[index_free].fd,
                        mbtk_ftp_ssl_read_buf_s,
                        sizeof(mbtk_ftp_ssl_read_buf_s),
                        60000,
                        &err_rw);
                printf("\nmbtk_sock_read:\n%s\n",mbtk_ftp_ssl_read_buf_s);

            mbtk_sock[handle]->infos[index_free].is_support_ssl=1;
        }else{
            mbtk_sock[handle]->infos[index_free].is_support_ssl=1;
        }
    }
#endif
    if(info->is_support_ssl){
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
    int ret = mbtk_sock_openssl_init(mbtk_sock[handle]->inter_infos[index_free].fd, info->ingnore_cert, &mbtk_sock[handle]->inter_infos[index_free].ssl_info);
    if(0 != ret){
        LOGE("mbtk_sock_openssl_init() fail");
        goto result_fail_with_close;
    }
    mbtk_sock[handle]->inter_infos[index_free].fd = mbtk_sock[handle]->inter_infos[index_free].ssl_info.fd;
#else
    LOGE("openssl nonsupport");
    goto result_fail_with_close;
#endif
    }

    *mbtk_errno = MBTK_SOCK_SUCCESS;

    mbtk_sock[handle]->sock_num++;
    return mbtk_sock[handle]->inter_infos[index_free].fd;
result_fail_with_close:
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
    mbtk_openssl_deinit(&mbtk_sock[handle]->inter_infos[index_free].ssl_info);
#endif

    close(mbtk_sock[handle]->inter_infos[index_free].fd);
    mbtk_sock[handle]->inter_infos[index_free].fd = -1;
result_fail:
    memset(&(mbtk_sock[handle]->inter_infos[index_free]),0x0,sizeof(mbtk_sock_inter_info_s));
    memset(&(mbtk_sock[handle]->infos[index_free]),0x0,sizeof(mbtk_sock_info));
    LOGE("mbtk_sock_open() end:fail");
    return -1;
}
extern int mbtk_ssl_init_func(mbtk_sock_handle handle ,bool ingnore_cert,mbtk_sock_session fd)
{
    int i=0;
    int index_free=0;

    for (i=0;i<10;i++)
    {
        if(mbtk_sock[handle]->inter_infos[i].fd == fd)
        {
            index_free = i;
            break;
        }
    }
    
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
    int ret = mbtk_sock_openssl_init(mbtk_sock[handle]->inter_infos[index_free].fd, ingnore_cert, &mbtk_sock[handle]->inter_infos[index_free].ssl_info);
    if(0 != ret){
        LOGE("mbtk_sock_openssl_init() fail");
        return -1;
    }
    mbtk_sock[handle]->inter_infos[index_free].fd = mbtk_sock[handle]->inter_infos[index_free].ssl_info.fd;
    return 0;
#else
    LOGE("openssl support");
	return -1;
#endif
}

extern int mbtk_ssl_close_func(mbtk_sock_handle handle ,bool ingnore_cert,mbtk_sock_session fd)
{
    int i=0;
    int index_free=0;

    for (i=0;i<10;i++)
    {
        if(mbtk_sock[handle]->inter_infos[i].fd == fd)
        {
            index_free = i;
            break;
        }
    }
    
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
    mbtk_openssl_deinit(&mbtk_sock[handle]->inter_infos[index_free].ssl_info);
    return 0;
#else
    LOGE("openssl nonsupport");
	return -1;
#endif
}

extern int mbtk_sock_write(mbtk_sock_handle handle,mbtk_sock_session session,
                const void *buffer,
                unsigned int buf_len,
                unsigned int timeout,
                int *mbtk_errno)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    *mbtk_errno = MBTK_SOCK_ERROR;
    if(buffer == NULL) {
        LOGE("mbtk_sock_write() args error.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }

    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    int len = 0;
    unsigned int count = 0;
    if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
        while(count < buf_len){
            if(mbtk_sock[handle]->infos[index].is_support_ssl) {
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
            len = mbtk_openssl_write(inter_info->ssl_info.ssl,(const unsigned char *)buffer + count,buf_len - count);
#else
            LOGE("openssl nonsupport");
		    return -1;
#endif
            } else
                len = write(inter_info->fd,(char*)buffer + count,buf_len - count);
            if(len < 0){
                if(errno == EWOULDBLOCK){
                    usleep(50000);
                    continue;
                } else {
                    LOGE("write error.[%d]",errno);
                    if(count <= 0)
                        count = -1;
                    break;
                }
            } else if(len == 0) {
                LOGE("write error(len == 0).[%d]",errno);
            } else {
                count += len;
            }
        }
    } else if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_UDP){
        // Start send data
        while(count < buf_len){
            len = sendto(inter_info->fd,(char*)buffer + count,buf_len - count,0,NULL,0);
            if(len < 0){
                if(errno == EWOULDBLOCK){
                    usleep(50000);
                    continue;
                } else {
                    LOGE("sendto error.[%d]",errno);
                    if(ECONNREFUSED == errno) { // Disconnected.
                        LOGD("Socket Disconnected.");
                    }
                    break;
                }
            } else if(len == 0) {
                LOGD("write error(len == 0).[%d]",errno);
            } else {
                count += len;
            }
        }
    } else {
        LOGE("Socket type error.");
        return -1;
    }

    if(count == buf_len){
        LOGD("Write data[%d/%d] success.",count,buf_len);
    } else { // Open session fail
        LOGD("Write data[%d/%d] fail.",count,buf_len);
    }

    *mbtk_errno = MBTK_SOCK_SUCCESS;
    return count;
}

extern int mbtk_sock_read(mbtk_sock_handle handle,mbtk_sock_session session,
            void *buffer,
            unsigned int buf_len,
            unsigned int timeout,
            int *mbtk_errno)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    *mbtk_errno = MBTK_SOCK_ERROR;
    if(buffer == NULL) {
        LOGE("mbtk_sock_write() args error.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }

    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    unsigned int count = 0;
    if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
        int len = 0;
        int try_count = 0;
        int times = timeout / 50;
        memset(buffer,0x0,buf_len);
        while(count < buf_len){
            try_count++;
            if(mbtk_sock[handle]->infos[index].is_support_ssl) {
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
            len = mbtk_openssl_read(inter_info->ssl_info.ssl,(unsigned char *)buffer + count,buf_len - count);
#else
            LOGE("openssl nonsupport");
		    return -1;
#endif			
            } else
                len = read(inter_info->fd,(char*)buffer + count,buf_len - count);
            if(len < 0){
                if(errno == EWOULDBLOCK){
                    if(count > 0) // Read data
                        break; // Read data end.

                    if(try_count >= times){ // Timeout
                        count = -1;
                        if(times != 0) {
                            *mbtk_errno = MBTK_SOCK_ETIMEOUT;
                        }
                        LOGE("Not read enough data,return.[%d/%d]",count,buf_len);
                        break;
                    } else {
                        usleep(50000);
                        continue;
                    }
                } else {
                    LOGE("read error.[%d]",errno);
                    if(count <= 0)
                        count = -1;
                    break;
                }
            } else if(len == 0) {
                LOGE("read error(len == 0).[%d]",errno);
                if(errno == EINPROGRESS) {
                    if(close(inter_info->fd) == 0) {// Success
                        LOGD("Socket disconnected.Close it.");
                    }
                    if(count <= 0)
                        count = -1;
                } else {
                    if(count <= 0)
                        count = 0;
                }
                break;
            } else {
                count += len;
            }
        }
    } else if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_UDP) {
        // Start recv data
        struct sockaddr_in seraddr;
        socklen_t seraddr_len;
        int try_count = 0;
        int times = timeout / 50;
        int len = 0;
        memset(buffer,0x0,buf_len);
        while(TRUE){
            try_count++;
            seraddr_len = sizeof(struct sockaddr_in);
            len = recvfrom(inter_info->fd,buffer,buf_len,0,(struct sockaddr*)&seraddr,&seraddr_len);
            if(len < 0){
                if(errno == EWOULDBLOCK){// No data can read.
                    if(count > 0) // Read data
                        break; // Read data end.

                    if(try_count >= times){ // Timeout
                        if(times == 0) {
                            LOGE("Can not read.");
                        } else {
                            LOGE("Timeout");
                            *mbtk_errno = MBTK_SOCK_ETIMEOUT;
                        }
                        count = -1;
                        LOGE("Not read enough data,return.[%d/%d]",count,buf_len);
                        break;
                    } else {
                        usleep(50000);
                        continue;
                    }
                } else {
                    LOGE("recvfrom error.[%d]",errno);
                    if(count <= 0)
                        count = -1;
                    break;
                }
            } else if(len == 0) {
                LOGE("write error(len == 0).[%d]",errno);
                if(count <= 0)
                    count = 0;
                break;
            } else {
                count += len;
            }
        }
    } else {
        LOGE("Socket type error.");
        return -1;
    }

//    if(count == buf_len){
//        LOGD("Read data[%d/%d] success.",count,buf_len);
//    } else { // Open session fail
//        LOGD("Read data[%d/%d] fail.",count,buf_len);
//    }

    LOGV("Read data[%d/%d].",count,buf_len);

    *mbtk_errno = MBTK_SOCK_SUCCESS;
    return count;
}
extern int mbtk_sock_readline(mbtk_sock_handle handle,mbtk_sock_session session,
            void *buffer,
            unsigned int buf_len,
            unsigned int timeout,
            int *mbtk_errno,
            int *read_line_count,
            char *buf_ptr)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    *mbtk_errno = MBTK_SOCK_ERROR;
    if(buffer == NULL) {
        LOGE("mbtk_sock_write() args error.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }

    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    unsigned int count = 0;
//    unsigned int read_count = 0;
    memset(buf_ptr, 0, buf_len);
    char *temp_ptr = (char *)buffer;
copy_angin_ssl:
    while(*read_line_count > 0 && *temp_ptr != '\n') {
        if(temp_ptr == NULL)
        {
            printf("\n*temp_ptr is null\n");
            goto read_end;
        }
        *buf_ptr++ = *temp_ptr++;
        (*read_line_count)--;
        count++;
    }
    if(*read_line_count == 0)
    {
        if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
            int len = 0;
            int try_count = 0;
            int times = timeout / 50;
            memset(buffer,0x0,buf_len);
            while(count < buf_len){
                try_count++;
                if(mbtk_sock[handle]->infos[index].is_support_ssl) {
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
                    len = mbtk_openssl_read(inter_info->ssl_info.ssl,(unsigned char*)buffer + count,buf_len - count);
#else
                    LOGE("openssl nonsupport");
					return -1;
#endif				
                } else
                    len = read(inter_info->fd,(char*)buffer + count,buf_len - count);
                *read_line_count = len;
                if(len < 0){
                    if(errno == EWOULDBLOCK){
                        if(count > 0) // Read data
                        {
                            *read_line_count = count;
                            count = 0;
                            goto copy_angin_ssl;
                            break; // Read data end.
                        }
                        else
                        {
                            //printf("\nread_end\n");
                            goto read_end;
                        }
                        if(try_count >= times){ // Timeout
                            count = -1;
                            if(times != 0) {
                                *mbtk_errno = MBTK_SOCK_ETIMEOUT;
                            }
                            LOGE("Not read enough data,return.[%d/%d]",count,buf_len);
                            goto read_end;
                            break;
                        } else {
                            usleep(50000);
                            continue;
                        }
                    } else {
                        LOGE("read error.[%d]",errno);
                        if(count <= 0)
                            count = -1;
                        else {
                            *read_line_count = count;
                        }
                        break;
                    }
                } else if(len == 0) {
                    LOGE("read error(len == 0).[%d]",errno);
                    if(errno == EINPROGRESS) {
                        if(close(inter_info->fd) == 0) {// Success
                            LOGD("Socket disconnected.Close it.");
                        }
                        if(count <= 0)
                            count = -1;
                    } else {
                        if(count <= 0)
                            count = 0;
                        else
                            count = -1;
                    }
                    goto read_end;
                    break;
                } else {
                    count += len;
                }
            }
        } else if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_UDP) {
            // Start recv data
            struct sockaddr_in seraddr;
            socklen_t seraddr_len;
            int try_count = 0;
            int times = timeout / 50;
            int len = 0;
            memset(buffer,0x0,buf_len);
            while(TRUE){
                try_count++;
                seraddr_len = sizeof(struct sockaddr_in);
                len = recvfrom(inter_info->fd,buffer,buf_len,0,(struct sockaddr*)&seraddr,&seraddr_len);
                if(len < 0){
                    if(errno == EWOULDBLOCK){// No data can read.
                        if(count > 0) // Read data
                            break; // Read data end.

                        if(try_count >= times){ // Timeout
                            if(times == 0) {
                                LOGE("Can not read.");
                                //printf("Can not read.\n");
                            } else {
                                LOGE("Timeout");
                                //printf("Timeout\n");
                                *mbtk_errno = MBTK_SOCK_ETIMEOUT;
                            }
                            count = -1;
                            LOGE("Not read enough data,return.[%d/%d]",count,buf_len);
                            //printf("Not read enough data,return.[%d/%d]\n",count,buf_len);
                            break;
                        } else {
                            usleep(50000);
                            continue;
                        }
                    } else {
                        LOGE("recvfrom error.[%d]",errno);
                        if(count <= 0)
                            count = -1;
                        break;
                    }
                } else if(len == 0) {
                    LOGE("write error(len == 0).[%d]",errno);
                    if(count <= 0)
                        count = 0;
                    break;
                } else {
                    count += len;
                }
            }
        } else {
            LOGE("Socket type error.");
            //printf("Socket type error.\n");
            return -1;
        }
        count = 0;
        goto copy_angin_ssl;
    } else if(*temp_ptr == '\n') { // Read line.
        *buf_ptr++ = '\n';
        (*read_line_count)--;
        count++;

        if(*read_line_count > 0)
            memcpy(buffer, temp_ptr + 1, *read_line_count);
        return count;
    }
    LOGV("Read data[%d/%d].",count,buf_len);
read_end:
    *mbtk_errno = MBTK_SOCK_SUCCESS;
    return count;
}

extern int mbtk_sock_read_async(mbtk_sock_handle handle,mbtk_sock_session session,
            void *buffer,
            unsigned int buf_len)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    if(buffer == NULL) {
        LOGE("mbtk_sock_write() args error.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }
    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    int len = 0;
    int read_count = 0;
    if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
        memset(buffer,0x0,buf_len);
        while(read_count < buf_len) {
            if(mbtk_sock[handle]->infos[index].is_support_ssl) {
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
                len = mbtk_openssl_read(inter_info->ssl_info.ssl,(unsigned char*)buffer + read_count,buf_len - read_count);
#else
                LOGE("openssl nonsupport");
				return -1;
#endif
            } else
                len = read(inter_info->fd,(char*)buffer  + read_count,buf_len - read_count);

            if(len > 0) {
                read_count += len;
            } else {
                if(errno == EWOULDBLOCK) { // No data
                    break;
                } else {
                    LOGE("Will retry : len = %d, errno = %d", len, errno);
                }
            }
        }
    } else if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_UDP) {
        // Start recv data
        struct sockaddr_in seraddr;
        socklen_t seraddr_len;
        memset(buffer,0x0,buf_len);
        seraddr_len = sizeof(struct sockaddr_in);
        memset(buffer,0x0,buf_len);

        while(read_count < buf_len) {
            len = recvfrom(inter_info->fd,buffer + read_count,buf_len - read_count,0,(struct sockaddr*)&seraddr,&seraddr_len);

            if(len > 0) {
                read_count += len;
            } else {
                if(errno == EWOULDBLOCK) { // No data
                    break;
                } else {
                    LOGE("Will retry : len = %d, errno = %d", len, errno);
                }
            }
        }
    } else {
        LOGE("Socket type error.");
        return -1;
    }

    LOGV("Read data[%d/%d].",read_count,buf_len);

    return read_count;
}

extern int mbtk_sock_read_sync(mbtk_sock_handle handle,mbtk_sock_session session,
            void *buffer,
            unsigned int buf_len)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    if(buffer == NULL) {
        LOGE("mbtk_sock_write() args error.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }
    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    int len;
    if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
TCP_READ_AGAIN:
        memset(buffer,0x0,buf_len);
        if(mbtk_sock[handle]->infos[index].is_support_ssl) {
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
            len = mbtk_openssl_read(inter_info->ssl_info.ssl,(unsigned char*)buffer,buf_len);
#else
            LOGE("openssl nonsupport");
			return -1;
#endif
        } else
            len = read(inter_info->fd,(char*)buffer,buf_len);
        if(len < 0){
            if(errno == EWOULDBLOCK){
                usleep(100000);
                LOGW("Read retry...");
                goto TCP_READ_AGAIN;
            } else {
                LOGE("read error.[%d]",errno);
                return -1;
            }
        }
    } else if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_UDP) {
        // Start recv data
        struct sockaddr_in seraddr;
        socklen_t seraddr_len;
UDP_READ_AGAIN:
        memset(buffer,0x0,buf_len);
        seraddr_len = sizeof(struct sockaddr_in);
        len = recvfrom(inter_info->fd,buffer,buf_len,0,(struct sockaddr*)&seraddr,&seraddr_len);
        if(len < 0){
            if(errno == EWOULDBLOCK){
                usleep(100000);
                goto UDP_READ_AGAIN;
            } else {
                LOGE("read error.[%d]",errno);
                return -1;
            }
        }
    } else {
        LOGE("Socket type error.");
        return -1;
    }

    LOGV("Read data[%d/%d].",len,buf_len);

    return len;
}


extern int mbtk_sock_close(mbtk_sock_handle handle,mbtk_sock_session session,
            unsigned int timeout,
            int *mbtk_errno)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    *mbtk_errno = MBTK_SOCK_ERROR;
    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session == mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }
    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

    int i;
    for(i = 0;i < MBTK_SOCK_MAX_NUM;i++) {
        if(mbtk_sock[handle]->inter_infos[i].fd == inter_info->fd){
            if(mbtk_sock[handle]->init_info.sock_cb) {
                struct epoll_event ev;
                ev.data.fd = inter_info->fd;
                ev.events = EPOLLIN | EPOLLET;
                epoll_ctl(epoll_fd,EPOLL_CTL_DEL,inter_info->fd,&ev);
            }

            if(close(inter_info->fd) < 0) {// Success
                LOGE("Close socket fail[%d].",errno);
                //break;
            }
            mbtk_sock[handle]->inter_infos[i].fd = -1;
            memset(&(mbtk_sock[handle]->infos[i]),0x0,sizeof(mbtk_sock_info));
            mbtk_sock[handle]->sock_num--;
            break;
        }
    }

    if(mbtk_sock[handle]->infos[index].is_support_ssl){
#ifdef MBTK_OPENSSL_V3_0_0_SUPPORT
            mbtk_openssl_deinit(&inter_info->ssl_info);
#endif
    }

    *mbtk_errno = MBTK_SOCK_SUCCESS;
    return 0;
}

extern int mbtk_sock_deinit(mbtk_sock_handle handle)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    if(mbtk_sock[handle]->sock_num > 0) {
        LOGE("There are socket not close.");
        return MBTK_SOCK_ERROR;
    }

    LOGD("mbtk_sock_deinit() start.");
#if 0
    sock_thread_running = FALSE;
    write(pipe_fds[0],"0",1);

    // Wait for thread exist.
    while(sock_inited) {
        usleep(100);
    }
#endif

    int i;
    for(i = 0;i < MBTK_SOCK_MAX_NUM;i++) {
        if(mbtk_sock[handle]->inter_infos[i].fd > 0){
            if(mbtk_sock[handle]->init_info.sock_cb) {
                struct epoll_event ev;
                ev.data.fd = mbtk_sock[handle]->inter_infos[i].fd;
                ev.events = EPOLLIN | EPOLLET;
                epoll_ctl(epoll_fd,EPOLL_CTL_DEL,mbtk_sock[handle]->inter_infos[i].fd,&ev);
            }

            if(close(mbtk_sock[handle]->inter_infos[i].fd) < 0) {// Success
                LOGE("Close socket fail[%d].",errno);
                //break;
            }
            mbtk_sock[handle]->inter_infos[i].fd = -1;
            memset(&(mbtk_sock[handle]->infos[i]),0x0,sizeof(mbtk_sock_info));
            break;
        }
    }

    //memset(&mbtk_sock,0x0,sizeof(mbtk_sock_s));
    free(mbtk_sock[handle]);
    mbtk_sock[handle] = NULL;
    LOGD("mbtk_sock_deinit() end.");
    return MBTK_SOCK_SUCCESS;
}

/*
* Get TCP RECV buffer data length.
*/
int mbtk_sock_tcp_recv_len_get(mbtk_sock_handle handle,mbtk_sock_session session)
{
    if(handle < 0 || handle >= MBTK_HANDLE_MAX_NUM
        || session < 0 || mbtk_sock[handle] == NULL) {
        LOGE("Socket not inited.");
        return -1;
    }

    mbtk_sock_inter_info_s *inter_info = NULL;
    int index = 0;
    while(index < MBTK_SOCK_MAX_NUM) {
        if(session ==
            mbtk_sock[handle]->inter_infos[index].fd) {
            inter_info = &(mbtk_sock[handle]->inter_infos[index]);
            break;
        }
        index++;
    }
    if(!sock_info_check(handle,inter_info)) {
        LOGE("sock_info_check() fail.");
        return -1;
    }

    index = sock_info_find_by_fd(handle,inter_info->fd);
    if(index < 0) {
        LOGE("No such socket in session list.");
        return -1;
    }

//    unsigned int count = 0;
    int len = 0;
    if(mbtk_sock[handle]->infos[index].type == MBTK_SOCK_TCP) {
        if(ioctl(inter_info->fd, FIONREAD, &len))
        {
            LOGE("Get ioctl FIONREAD fail:%d", errno);
            return -1;
        }
    } else {
        LOGE("Only surrport for TCP.");
        return -1;
    }

    return len;
}

void mbtk_net_lib_info_print()
{
    MBTK_SOURCE_INFO_PRINT("mbtk_net_lib");
}