mbtk/libmbtk_lib_v2/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>

#ifdef MBTK_POLARSSL_SUPPORT
#include <polarssl/net.h>
#include <polarssl/ssl.h>
#include <polarssl/entropy.h>
#include <polarssl/ctr_drbg.h>
#include <polarssl/certs.h>
#include <polarssl/x509.h>
#include <polarssl/error.h>
#include <polarssl/debug.h>
#include <polarssl/config.h>
#else
#include <resolv.h>
#include <openssl/ssl.h>
#include <openssl/err.h>

//#define SSL_VERIFY_PEER                 0x01
//#define SSL_FILETYPE_PEM                 0x01
//#define SSL_VERIFY_FAIL_IF_NO_PEER_CERT 0x02

#define DFL_CA_FILE             "/ca.crt"
#define DFL_CRT_FILE            "/client.crt"
#define DFL_KEY_FILE            "/client.key"
#endif
#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_openssl.h"

#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 = -1;
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;
}

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;
}

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;
                            int 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++;
        }
    }
}

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;
}

#ifdef MBTK_POLARSSL_SUPPORT
static int mbtk_polarssl_open(int fd ,bool ingnore_cert,mbtk_sock_inter_info_s* inter_info)
{
    LOGE("8\n");
    int ret = 0, len, tail_len, i, written, frags;
    unsigned char buf[SSL_MAX_CONTENT_LEN + 1];
    const char *pers = "ssl_client";
    opt.server_name         = DFL_SERVER_NAME;
    opt.server_addr         = DFL_SERVER_ADDR;
    opt.server_port         = DFL_SERVER_PORT;
    opt.debug_level         = DFL_DEBUG_LEVEL;
    opt.nbio                = DFL_NBIO;
    opt.request_page        = DFL_REQUEST_PAGE;
    opt.request_size        = DFL_REQUEST_SIZE;
    opt.ca_file             = DFL_CA_FILE;
    opt.ca_path             = DFL_CA_PATH;
    opt.crt_file            = DFL_CRT_FILE;
    opt.key_file            = DFL_KEY_FILE;
    opt.psk                 = DFL_PSK;
    opt.psk_identity        = DFL_PSK_IDENTITY;
    opt.force_ciphersuite[0]= DFL_FORCE_CIPHER;
    opt.renegotiation       = DFL_RENEGOTIATION;
    opt.allow_legacy        = DFL_ALLOW_LEGACY;
    opt.renegotiate         = DFL_RENEGOTIATE;
    opt.exchanges           = DFL_EXCHANGES;
    opt.min_version         = DFL_MIN_VERSION;
    opt.max_version         = DFL_MAX_VERSION;
    opt.auth_mode           = DFL_AUTH_MODE;
    opt.mfl_code            = DFL_MFL_CODE;
    opt.trunc_hmac          = DFL_TRUNC_HMAC;
    opt.reconnect           = DFL_RECONNECT;
    opt.reco_delay          = DFL_RECO_DELAY;
    opt.tickets             = DFL_TICKETS;
    opt.alpn_string         = DFL_ALPN_STRING;

    entropy_context entropy;
    ctr_drbg_context ctr_drbg;
    ssl_context ssl;
    ssl_session saved_session;
    x509_crt cacert;
    x509_crt clicert;
    pk_context pkey;

    memset( &ssl, 0, sizeof( ssl_context ) );
    memset( &saved_session, 0, sizeof( ssl_session ) );
    x509_crt_init( &cacert );
    x509_crt_init( &clicert );
    pk_init( &pkey );
    LOGE("9\n");
     /*
     * 0. Initialize the RNG and the session data
     */

    entropy_init( &entropy );
    if( ( ret = ctr_drbg_init( &ctr_drbg, entropy_func, &entropy,
                               (const unsigned char *) pers,
                               strlen( pers ) ) ) != 0 )
    {
        LOGE( " failed\n  ! ctr_drbg_init returned -0x%x\n", -ret );
        return -1;
    }
    if(!ingnore_cert)
    {
        LOGE("10\n");
     /*
     * 1.1. Load the trusted CA
     */
    //ret = x509_crt_parse(&cacert,ca1_cert,strlen(ca1_cert));
        ret = x509_crt_parse_file( &cacert, opt.ca_path );
        if( ret < 0 )
        {
            LOGE( " failed\n  !  ca x509_crt_parse returned -0x%x\n\n", -ret );
            return -1;
        }

         /*
         * 1.2. Load own certificate and private key
         *
         * (can be skipped if client authentication is not required)
         */

        ret = x509_crt_parse_file( &clicert, opt.crt_file );
        if( ret != 0 )
        {
            LOGE( " failed\n  !  crt x509_crt_parse returned -0x%x\n\n", -ret );
            return -1;
        }

        ret = pk_parse_keyfile( &pkey, opt.key_file, NULL);
        if( ret != 0 )
        {
            LOGE( " failed\n  !  key x509_crt_parse returned -0x%x\n\n", -ret );
            return -1;
        }
    }
     /*
     * 2. Setup stuff
     */
    LOGE( "  . Setting up the SSL/TLS structure..." );

    if( ( ret = ssl_init( &ssl ) ) != 0 )
    {
        LOGE( " failed\n  ! ssl_init returned -0x%x\n\n", -ret );
        return -1;
    }

    ssl_set_endpoint( &ssl, SSL_IS_CLIENT );
    if(ingnore_cert)
    {
        opt.auth_mode = SSL_VERIFY_OPTIONAL;
    }
    else
    {
        opt.auth_mode = SSL_VERIFY_REQUIRED;
    }

    ssl_set_authmode( &ssl, opt.auth_mode );

    ssl_set_rng( &ssl, ctr_drbg_random, &ctr_drbg );

    ssl_set_bio( &ssl, net_recv, &fd, net_send, &fd );

    ssl_set_renegotiation( &ssl, opt.renegotiation );
    ssl_legacy_renegotiation( &ssl, opt.allow_legacy );

    ssl_set_ca_chain( &ssl, &cacert, NULL, NULL );
    if(!ingnore_cert)
    {
        if( ( ret = ssl_set_own_cert( &ssl, &clicert, &pkey ) ) != 0 )
            {
                LOGE( " failed\n  ! ssl_set_own_cert returned %d\n\n", ret );
                    return -1;
            }
    }
    if( opt.min_version != -1 )
        ssl_set_min_version( &ssl, SSL_MAJOR_VERSION_3, opt.min_version );
    if( opt.max_version != -1 )
        ssl_set_max_version( &ssl, SSL_MAJOR_VERSION_3, opt.max_version );
    /*
     * 3. Handshake
     */
    LOGE( "  . Performing the SSL/TLS handshake..." );

    while( ( ret = ssl_handshake( &ssl ) ) != 0 )
    {
        if( ret != POLARSSL_ERR_NET_WANT_READ && ret != POLARSSL_ERR_NET_WANT_WRITE )
        {
            LOGE( " failed\n  ! ssl_handshake returned -0x%x\n", -ret );
            if( ret == POLARSSL_ERR_X509_CERT_VERIFY_FAILED )
                LOGE(
                    "    Unable to verify the server's certificate. "
                        "Either it is invalid,\n"
                    "    or you didn't set ca_file or ca_path "
                        "to an appropriate value.\n"
                    "    Alternatively, you may want to use "
                        "auth_mode=optional for testing purposes.\n" );
            LOGE( "\n" );
            return -1;;
        }
    }

    LOGE( " ok\n    [ Protocol is %s ]\n    [ Ciphersuite is %s ]\n",ssl_get_version( &ssl ), ssl_get_ciphersuite( &ssl ) );
    printf( " ok\n    [ Protocol is %s ]\n    [ Ciphersuite is %s ]\n",ssl_get_version( &ssl ), ssl_get_ciphersuite( &ssl ) );

    /*
     * 4. Verify the server certificate
     */
    LOGE( "  . Verifying peer X.509 certificate..." );

    if( ( ret = ssl_get_verify_result( &ssl ) ) != 0 )
    {
        LOGE( " failed\n" );

        if( ( ret & BADCERT_EXPIRED ) != 0 )
            LOGE( "  ! server certificate has expired\n" );

        if( ( ret & BADCERT_REVOKED ) != 0 )
            LOGE( "  ! server certificate has been revoked\n" );

        if( ( ret & BADCERT_CN_MISMATCH ) != 0 )
            LOGE( "  ! CN mismatch (expected CN=%s)\n", opt.server_name );

        if( ( ret & BADCERT_NOT_TRUSTED ) != 0 )
            LOGE( "  ! self-signed or not signed by a trusted CA\n" );

    }

    if( ssl_get_peer_cert( &ssl ) != NULL )
    {
        LOGE( "  . Peer certificate information    ...\n" );
        x509_crt_info( (char *) buf, sizeof( buf ) - 1, "      ",
                       ssl_get_peer_cert( &ssl ) );
        LOGE( "%s\n", buf );
    }

    inter_info->cacert = &cacert;
    inter_info->clicert = &clicert;
    inter_info->ctr_drbg = &ctr_drbg;
    inter_info->entropy = &entropy;
    inter_info->pkey = &pkey;
    inter_info->saved_session = &saved_session;
    inter_info->ssl  = &ssl;

    return 0;
}

static int mbtk_polarssl_close(mbtk_sock_inter_info_s *inter_info)
{
    if (inter_info == NULL)
    {
        return -1;
    }

    int ret = -1;
    while( ( ret = ssl_close_notify( inter_info->ssl ) ) < 0 )
    {
        if( ret == POLARSSL_ERR_NET_CONN_RESET )
        {
            LOGE( " ok (already closed by peer)\n" );
            ret = 0;
            return -1;
        }

        if( ret != POLARSSL_ERR_NET_WANT_READ &&
            ret != POLARSSL_ERR_NET_WANT_WRITE )
        {
            LOGE( " failed\n  ! ssl_close_notify returned %d\n\n", ret );
            return -1;
        }
    }

    x509_crt_free( inter_info->clicert );
    x509_crt_free( inter_info->cacert );
    pk_free( inter_info->pkey );
    ssl_session_free( inter_info->saved_session );
    ssl_free( inter_info->ssl );
    ctr_drbg_free( inter_info->ctr_drbg );
    entropy_free( inter_info->entropy );
    return 0;
}

static int mbtk_polarssl_write( ssl_context *ssl, const unsigned char *buf, size_t len )
{
    return ssl_write(ssl, buf, len);
}

static int mbtk_polarssl_read( ssl_context *ssl, unsigned char *buf, size_t len )
{
    return ssl_read(ssl, buf, len);
}

#else

void ShowCerts(SSL * ssl)
{
    X509 *cert;
    char *line;

    cert = SSL_get_peer_certificate(ssl);
    // SSL_get_verify_result()是重点，SSL_CTX_set_verify()只是配置启不启用并没有执行认证，调用该函数才会真证进行证书认证
    // 如果验证不通过，那么程序抛出异常中止连接
    if(SSL_get_verify_result(ssl) == X509_V_OK){
        printf("证书验证通过\n");
    }
    if (cert != NULL) {
        printf("数字证书信息:\n");
        line = X509_NAME_oneline(X509_get_subject_name(cert), 0, 0);
        printf("证书: %s\n", line);
        free(line);
        line = X509_NAME_oneline(X509_get_issuer_name(cert), 0, 0);
        printf("颁发者: %s\n", line);
        free(line);
        X509_free(cert);
    } else
        printf("无证书信息！\n");
}

static int mbtk_openssl_open(int fd ,bool ingnore_cert,mbtk_sock_inter_info_s* inter_info)
{
    SSL_CTX *ctx;
    SSL *ssl;

    /* SSL 库初始化，参看 ssl-server.c 代码 */
    SSL_library_init();
    OpenSSL_add_all_algorithms();
    SSL_load_error_strings();
    ctx = SSL_CTX_new(SSLv23_client_method());
    if (ctx == NULL) {
        ERR_print_errors_fp(stdout);
        return -1;
    }

    if(!ingnore_cert)
    {
        // 双向验证
        // SSL_VERIFY_PEER---要求对证书进行认证，没有证书也会放行
        // SSL_VERIFY_FAIL_IF_NO_PEER_CERT---要求客户端需要提供证书，但验证发现单独使用没有证书也会放行
        SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER|SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);
        // 设置信任根证书
        if (SSL_CTX_load_verify_locations(ctx, "/ca.crt",NULL)<=0){
            ERR_print_errors_fp(stdout);
             printf("fail SSL_CTX_load_verify_locations()\n");
            return -1;
        }

        /* 载入用户的数字证书， 此证书用来发送给客户端。 证书里包含有公钥 */
        if (SSL_CTX_use_certificate_file(ctx, DFL_CRT_FILE, SSL_FILETYPE_PEM) <= 0) {
            ERR_print_errors_fp(stdout);
            printf("fail SSL_CTX_use_certificate_file()\n");
            return -1;
        }
        /* 载入用户私钥 */
        if (SSL_CTX_use_PrivateKey_file(ctx, DFL_KEY_FILE, SSL_FILETYPE_PEM) <= 0) {
            ERR_print_errors_fp(stdout);
            printf("fail SSL_CTX_use_PrivateKey_file()\n");
            return -1;
        }
        /* 检查用户私钥是否正确 */
        if (!SSL_CTX_check_private_key(ctx)) {
            ERR_print_errors_fp(stdout);
             printf("fail SSL_CTX_check_private_key()\n");
            return -1;
        }

    }

    /* 基于 ctx 产生一个新的 SSL */
    ssl = SSL_new(ctx);
    SSL_set_fd(ssl, fd);
    /* 建立 SSL 连接 */
    if (SSL_connect(ssl) == -1)
        ERR_print_errors_fp(stderr);
    else {
        printf("Connected with %s encryption\n", SSL_get_cipher(ssl));
        if(!ingnore_cert)
        {
            ShowCerts(ssl);
        }
    }

    inter_info->ctx = &ctx;

    inter_info->ssl = &ssl;

    return 0;
}

static int mbtk_openssl_close(mbtk_sock_inter_info_s *inter_info)
{
    SSL_shutdown(inter_info->ssl);
    SSL_free(inter_info->ssl);
//    close(sockfd);
    SSL_CTX_free(inter_info->ctx);
    return 0;
}

static int mbtk_openssl_write( SSL *ssl, const unsigned char *buf, size_t len )
{
    return SSL_write(ssl, buf, len);
}

static int mbtk_openssl_read( SSL *ssl, unsigned char *buf, size_t len )
{
    return SSL_read(ssl, buf, len);
}

#endif

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;
                int 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,code;
            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_POLARSSL_SUPPORT
        if(mbtk_polarssl_open(mbtk_sock[handle]->inter_infos[index_free].fd,info->ingnore_cert,&mbtk_sock[handle]->inter_infos[index_free]) == -1){
            LOGE("mbtk_openssl_init fail");
            goto result_fail_with_close;
        }
#else
    if(mbtk_openssl_open(mbtk_sock[handle]->inter_infos[index_free].fd,info->ingnore_cert,&mbtk_sock[handle]->inter_infos[index_free]) == -1){
        LOGE("mbtk_openssl_init fail");
        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:
    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_POLARSSL_SUPPORT
    return mbtk_polarssl_open(mbtk_sock[handle]->inter_infos[index_free].fd,ingnore_cert,&mbtk_sock[handle]->inter_infos[index_free]);
#else
    return mbtk_openssl_open(mbtk_sock[handle]->inter_infos[index_free].fd,ingnore_cert,&mbtk_sock[handle]->inter_infos[index_free]);
#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_POLARSSL_SUPPORT
    if(mbtk_sock[handle]->inter_infos[index_free].ssl!=NULL);
        printf("\nmbtk_sock[handle]->inter_infos[index_free].ssl not empty\n");
    return mbtk_polarssl_close(&mbtk_sock[handle]->inter_infos[index_free]);
#else
    if(mbtk_sock[handle]->inter_infos[index_free].ssl!=NULL);
        printf("\nmbtk_sock[handle]->inter_infos[index_free].ssl not empty\n");
    return mbtk_openssl_close(&mbtk_sock[handle]->inter_infos[index_free]);
#endif
}

extern int mbtk_sock_write(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;
    }

    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_POLARSSL_SUPPORT
                len = mbtk_polarssl_write(inter_info->ssl,(char*)buffer + count,buf_len - count);
#else
            len = mbtk_openssl_write(inter_info->ssl,(char*)buffer + count,buf_len - count);

#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_POLARSSL_SUPPORT
                len = mbtk_polarssl_read(inter_info->ssl,(char*)buffer + count,buf_len - count);
#else
            len = mbtk_openssl_read(inter_info->ssl,(char*)buffer + count,buf_len - count);

#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,&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_POLARSSL_SUPPORT
                    len = mbtk_polarssl_read(inter_info->ssl,(char*)buffer + count,buf_len - count);
#else
                    len = mbtk_openssl_read(inter_info->ssl,(char*)buffer + count,buf_len - count);
#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,&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_POLARSSL_SUPPORT
                len = ssl_read(inter_info->ssl,(char*)buffer + read_count,buf_len - read_count);
#else
                len = mbtk_openssl_read(inter_info->ssl,(char*)buffer + read_count,buf_len - read_count);

#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,&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_POLARSSL_SUPPORT
            len = ssl_read(inter_info->ssl,(char*)buffer,buf_len);
#else
            len = mbtk_openssl_read(inter_info->ssl,(char*)buffer,buf_len);

#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,&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_POLARSSL_SUPPORT
        if(mbtk_polarssl_close(inter_info)== -1)
        {
            LOGE("close ssl fail");
            return -1;
        }
#else
    if(mbtk_openssl_close(inter_info)== -1)
    {
        LOGE("close ssl fail");
        return -1;
    }

#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");
}