mbtk/mbtk_rtpd/src/rtp_server.c

/*
*    rtp_server.c
*
*    RTP transport service.
*
*                  -------------------------
*                 |                         |
*    ------  UDP  |                -------  |
*   |      | -----|-> playback -> |       | |
*   | MCU  |      |               | Voice | |
*   |      |  UDP |               |       | |
*    ------  <----|-- record <--- |       | |
*                 |                -------  |
*                  -------------------------
*/
/******************************************************************************

                          EDIT HISTORY FOR FILE

  WHEN        WHO       WHAT,WHERE,WHY
--------    --------    -------------------------------------------------------
2024/12/3     LiuBin      Initial version

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

#include "mbtk_log.h"
#include "mbtk_utils.h"
#include "mbtk_audio2.h"
#include "mbtk_rtp_internal.h"
#include "mbtk_ifc.h"

#define RTP_DEBUG_PCM_FILE 1
#define RTP_UDP_READ_BUFF_SIZE 2048

static rtp_udp_ser_state_enum udp_state = RTP_UDP_SER_STATE_IDEL;
static rtp_voip_ser_state_enum voip_state = RTP_VOIP_SER_STATE_IDEL;
extern rtp_info_t rtp_info;
static pthread_t voip_playback;

static rtp_header_info_t rtp_head = {
    .version = 2,     // 2 (2 bit)
    .padding = 0,     // 0 (1 bit)
    .extension = 0,   // 0 (1 bit)
    .csrc = 1,        // 1 (4 bit)
    .marker = 0,      // 0 (1 bit)
    .payload_type = 0x60, // 0x60 (7 bit)
    .sequence = 0,    // (16 bit)
    .timestamp = 0,   // (32 bit)
    .ssrc = 0xFFFF0000,    // 0xFFFF0000 (32 bit)
    .csrc = 0xFFFF0000     // 0xFFFF0000 (32 bit)
};

int epoll_fd_add(int fd);
int epoll_fd_del(int fd);

static void *mbtk_memcpy(const void *src, unsigned int n)
{
    void *dest = malloc(n);
    if(dest) {
        return memcpy(dest, src, n);
    } else {
        return NULL;
    }
}

static void rtp_send_init()
{
    rtp_info.send.iov_idx = 1;      // First is RTP header.
    rtp_info.send.pack_size = 0;
    rtp_info.send.remain_buff_len = 0;
    rtp_info.recv.first_packet = TRUE;

    if(rtp_info.recv.recv_buff == NULL)
        rtp_info.recv.recv_buff = mbtk_loopbuff_get(1024 * 1024);

    rtp_head.sequence = (uint32) (rand()*rand());
    rtp_head.timestamp = 0;
}

static int rtp_pack_push(const uint8 *data, uint32 data_len, uint32_t timestamp)
{
    UNUSED(data);
    UNUSED(data_len);
    UNUSED(timestamp);

    if(rtp_info.recv.recv_buff) {
        if(data_len != mbtk_loopbuff_write(rtp_info.recv.recv_buff, data, data_len))
            return -1;
    }
    return 0;
}

static void udp_read_cb(int fd)
{
    int size;
    size_t audio_length;
    size_t metadata_length;
    struct msghdr m;
    //struct cmsghdr *cm;
    struct iovec iov;
    uint32_t header;
    uint32_t ssrc;
    uint8_t payload;
    unsigned cc;
    ssize_t r;
    uint8_t aux[1024];
    //bool found_tstamp = false;
    uint32_t timestamp;
    int64_t k, j, delta;

    uint8_t recv_buf[RTP_UDP_READ_BUFF_SIZE];
    //size_t recv_buf_size = 0;

    if (ioctl(fd, FIONREAD, &size) < 0) {
        LOGE("FIONREAD failed: %d", errno);
        goto fail;
    }

    if (size <= 0) {
        /* size can be 0 due to any of the following reasons:
         *
         * 1. Somebody sent us a perfectly valid zero-length UDP packet.
         * 2. Somebody sent us a UDP packet with a bad CRC.
         *
         * It is unknown whether size can actually be less than zero.
         *
         * In the first case, the packet has to be read out, otherwise the
         * kernel will tell us again and again about it, thus preventing
         * reception of any further packets. So let's just read it out
         * now and discard it later, when comparing the number of bytes
         * received (0) with the number of bytes wanted (1, see below).
         *
         * In the second case, recvmsg() will fail, thus allowing us to
         * return the error.
         *
         * Just to avoid passing zero-sized memchunks and NULL pointers to
         * recvmsg(), let's force allocation of at least one byte by setting
         * size to 1.
         */
        size = 1;
    }

    iov.iov_base = recv_buf;
    iov.iov_len = (size_t) size;

    m.msg_name = NULL;
    m.msg_namelen = 0;
    m.msg_iov = &iov;
    m.msg_iovlen = 1;
    m.msg_control = aux;
    m.msg_controllen = sizeof(aux);
    m.msg_flags = 0;

    r = recvmsg(fd, &m, 0);

    if (r != size) {
        if (r < 0 && errno != EAGAIN && errno != EINTR)
            LOGE("recvmsg() failed: %s", r < 0 ? strerror(errno) : "size mismatch");

        goto fail;
    }

    if(voip_state != RTP_VOIP_SER_STATE_RUNNING) {
        return;
    }

#if 0
    printf("RECV : %d\n", r);

    int send_len = mbtk_audio_pcm_play_data_send(recv_buf, r);
    if(r != send_len) {
        printf("play_data_send fail: %d/%d\n", send_len, r);
    }
#endif
    if (size < 12) {
        LOGE("RTP packet too short.");
        goto fail;
    }

#if 1
    memcpy(&header, iov.iov_base, sizeof(uint32_t));
    memcpy(&timestamp, (uint8_t*) iov.iov_base + 4, sizeof(uint32_t));
    memcpy(&ssrc, (uint8_t*) iov.iov_base + 8, sizeof(uint32_t));

    header = ntohl(header);
    timestamp = ntohl(timestamp);
    ssrc = ntohl(ssrc);

    if ((header >> 30) != 2) {
        LOGE("Unsupported RTP version.");
        goto fail;
    }

    if ((header >> 29) & 1) {
        LOGE("RTP padding not supported.");
        goto fail;
    }

    if ((header >> 28) & 1) {
        LOGE("RTP header extensions not supported.");
        goto fail;
    }

    if (ssrc != rtp_head.ssrc) {
        LOGE("Got unexpected SSRC");
        goto fail;
    }

    cc = (header >> 24) & 0xF;
    payload = (uint8_t) ((header >> 16) & 127U);
    rtp_head.sequence = (uint16_t) (header & 0xFFFFU);

    metadata_length = 12 + cc * 4;

    if (payload != rtp_head.payload_type) {
        LOGE("Got unexpected payload: %u", payload);
        goto fail;
    }

    if (metadata_length > (unsigned) size) {
        LOGE("RTP packet too short. (CSRC)");
        goto fail;
    }

    audio_length = size - metadata_length;

    if (audio_length % rtp_info.frame_size != 0) {
        LOGE("Bad RTP packet size.");
        goto fail;
    }

    if (rtp_info.recv.first_packet) {
        rtp_info.recv.first_packet = FALSE;
        rtp_info.recv.offset = timestamp;
    }

    /* Check whether there was a timestamp overflow */
    k = (int64_t) timestamp - (int64_t) rtp_info.recv.offset;
    j = (int64_t) 0x100000000LL - (int64_t) rtp_info.recv.offset + (int64_t) timestamp;

    if ((k < 0 ? -k : k) < (j < 0 ? -j : j))
        delta = k;
    else
        delta = j;

    LOGD("RECV : %d, delta = %d", audio_length, delta);
//     pa_memblockq_seek(s->memblockq, delta * (int64_t) rtp_info.frame_size, PA_SEEK_RELATIVE, true);

    /* The next timestamp we expect */
    rtp_info.recv.offset = timestamp + (uint32_t) (audio_length / rtp_info.frame_size);


    if(rtp_pack_push(recv_buf + metadata_length, audio_length, timestamp)) {
        LOGE("rtp_pack_push() fail.");
    }

#endif

    return;

fail:

    return;
}

static void voip_recorder_cb(void *data, uint32 data_len)
{
    if(data_len > 0) {
        LOGD("Record : %d", data_len);
        if(rtp_info.udp_send_sock.fd > 0) {
#if 0
            int len = sendto(rtp_info.udp_send_sock.fd, data, data_len, 0, NULL, 0);
            printf("SEND : %d / %d\n", len, data_len);
#else
            // 有剩余数据
            if(rtp_info.send.iov_idx == 1 && rtp_info.send.remain_buff_len > 0) {
                rtp_info.send.iov[rtp_info.send.iov_idx].iov_base = mbtk_memcpy(rtp_info.send.remain_buff,
                            rtp_info.send.remain_buff_len);
                rtp_info.send.iov[rtp_info.send.iov_idx].iov_len = rtp_info.send.remain_buff_len;
                rtp_info.send.iov_idx++;

                rtp_info.send.pack_size += rtp_info.send.remain_buff_len;
                rtp_info.send.remain_buff_len = 0;
            }

            // UDP各分包总大小不超过 c->mtu （           默认: DEFAULT_MTU ）
            // k 为分包大小
            uint32 k = rtp_info.send.pack_size + data_len > rtp_info.send.mtu ?
                            rtp_info.send.mtu - rtp_info.send.pack_size : data_len;

            rtp_info.send.iov[rtp_info.send.iov_idx].iov_base = mbtk_memcpy(data, k);
            rtp_info.send.iov[rtp_info.send.iov_idx].iov_len = k;
            rtp_info.send.iov_idx++;

            rtp_info.send.pack_size += k;

            if(rtp_info.send.pack_size % rtp_info.frame_size != 0) {
                LOGW("pack size error: %d - %d", rtp_info.send.pack_size, rtp_info.frame_size);
                return;
            }

            if (rtp_info.send.pack_size >= rtp_info.send.mtu || rtp_info.send.iov_idx >= MAX_IOVECS) {
                uint32_t header[4];
                struct msghdr m;
                ssize_t k;
                int i;

                header[0] = htonl(((uint32_t) rtp_head.version << 30) | ((uint32_t) rtp_head.csrc_count << 24) | ((uint32_t) rtp_head.payload_type << 16) | ((uint32_t) rtp_head.sequence));
                header[1] = htonl(rtp_head.timestamp);
                header[2] = htonl(rtp_head.ssrc);
                header[3] = htonl(rtp_head.csrc);

                rtp_info.send.iov[0].iov_base = (void*)header;
                rtp_info.send.iov[0].iov_len = sizeof(header);

                m.msg_name = NULL;
                m.msg_namelen = 0;
                m.msg_iov = rtp_info.send.iov;
                m.msg_iovlen = (size_t) rtp_info.send.iov_idx;
                m.msg_control = NULL;
                m.msg_controllen = 0;
                m.msg_flags = 0;

                k = sendmsg(rtp_info.udp_send_sock.fd, &m, MSG_DONTWAIT);

                for (i = 1; i < rtp_info.send.iov_idx; i++) {
                    free(rtp_info.send.iov[i].iov_base);
                    rtp_info.send.iov[i].iov_base = NULL;
                }

                rtp_head.sequence++;

                // 时间单位转为帧数（每帧多少us）
                rtp_head.timestamp += (unsigned) (rtp_info.send.pack_size / rtp_info.frame_size);

                if (k < 0) {
                    if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */
                        LOGE("sendmsg() failed: %s", strerror(errno));
                    return;
                }

                rtp_info.send.pack_size = 0;
                rtp_info.send.iov_idx = 1;
            }
#endif
        }
    } else {
        LOGD("Recorver data end.");
    }
}

static int64 time_us_get()
{
    struct timespec ts;
    memset(&ts, 0, sizeof(struct timespec));

    if(clock_gettime(CLOCK_REALTIME, &ts)) {
        LOGE("clock_gettime() fail:%d", errno);
        return -1;
    }

    return ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
}

int ip_get_from_dev(const char *dev_name, void *addr)
{
    int ret = 0;
    if(mbtk_ifc_open()) {
        LOGE("mbtk_ifc_open() fail.");
        ret = -1;
        goto return_result;
    }

    if(mbtk_ifc_get_addr(dev_name, addr)) {
        LOGE("mbtk_ifc_get_addr() fail.");
        ret = -1;
        goto return_result;
    }

return_result:
    mbtk_ifc_close();
    return ret;
}

static void voip_playback_thread(void *arg)
{
    UNUSED(arg);
    char *buff = (char*)malloc(rtp_info.playback_size * 2);
    if(buff == NULL) {
        LOGE("malloc() fail.");
        return;
    }

    if(mbtk_audio_voice_pcm_playback_start()) {
        LOGE("mbtk_audio_voice_pcm_playback_start() fail.");
        return;
    }

    usleep(100000);

    int64 play_start = time_us_get();   // us
    //uint64 play_count = 0;
    //char num_buff[1024] = {0};
    while (voip_state == RTP_VOIP_SER_STATE_RUNNING) {
        int len;
        if((len = mbtk_loopbuff_read(rtp_info.recv.recv_buff, buff, rtp_info.playback_size * 2)) > 0) {
            int send_len = mbtk_audio_pcm_play_data_send(buff, len);
            if(send_len > 0) {
                if(len != send_len) {
                    LOGW("play_data_send fail: %d/%d\n", send_len, len);
                }
                //play_count += send_len;
                int64 play_now = time_us_get();
                int time_offset = (int)((play_now - play_start) / 1000 * rtp_info.sample_for_ms);
                LOGD("loopbuff:%d, time_offset:%d, send_len:%d", mbtk_loopbuff_size(rtp_info.recv.recv_buff), time_offset,
                    send_len / 2);
#if 0
                int offset = send_len / 2 - time_offset;
                if(offset < 0 && len > -offset * 2) {
#if 0
                    if(mbtk_loopbuff_seek(rtp_info.recv.recv_buff, offset * 2)) {
                        LOGE("mbtk_loopbuff_seek() fail.");
                    }
                    mbtk_loopbuff_print(rtp_info.recv.recv_buff);
#else
                    mbtk_audio_pcm_play_data_send(buff + len + offset * 2, -offset * 2);
#endif
                }
#endif
                play_start = play_now;
            }
        }
#if 0
        else {
            if(voip_state == RTP_VOIP_SER_STATE_RUNNING)
                usleep(5000);
        }
        // LOGD("%s: No.%d frame playback.", __FUNCTION__, ++frames);

        int time_offset = (int)((time_us_get() - play_start) / 1000 * rtp_info.sample_for_ms);
        LOGD("loopbuff:%d, time_offset:%d, play_count:%d", mbtk_loopbuff_size(rtp_info.recv.recv_buff), time_offset,
            play_count / 2);
#endif
    }

    if(mbtk_audio_pcm_play_stop()) {
        LOGE("mbtk_audio_pcm_play_stop() fail.");
        return;
    }

    free(buff);
    return;
}


static int udp_setsockopt(int fd)
{
    int priority = 6;
    if (setsockopt(fd, SOL_SOCKET, SO_PRIORITY, (const void *) &priority, sizeof(priority)) < 0) {
        LOGE("setsockopt(SO_PRIORITY) fail:errno - %d", errno);
        return -1;
    }

    int one = 1;
    if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) {
        LOGE("SO_TIMESTAMP failed: %d", errno);
        return -1;
    }

    one = 1;
    if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) {
        LOGE("SO_REUSEADDR failed: %d", errno);
        return -1;
    }

    return 0;
}

static int socket_noblock(int fd)
{
    // Set O_NONBLOCK
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0)
    {
        LOGE("Get flags error:%d", errno);
        return -1;
    }
    flags |= O_NONBLOCK;
    if (fcntl(fd, F_SETFL, flags) < 0)
    {
        LOGE("Set flags error:%d", errno);
        return -1;
    }

    return 0;
}

static int rtp_udp_ser_open(const char *local_addr, int local_port)
{
    // No set local addr.
    UNUSED(local_addr);

    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if(fd < 0){
        LOGE("socket() fail.[%d]", errno);
        return -1;
    }

    if(udp_setsockopt(fd)) {
        goto result_fail_with_close;
    }

    struct sockaddr_in servaddr;
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(local_port);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    if (bind(fd, (struct sockaddr *)&servaddr, sizeof(struct sockaddr_in)) < 0) {
        LOGE("bind() failed: %d", errno);
        goto result_fail_with_close;
    }

    return fd;
result_fail_with_close:
    close(fd);
    fd = -1;
    LOGE("mbtk_sock_open() end:fail");
    return -1;
}

static int rtp_udp_cli_open(const char *vlan, const char *remote_addr, int remote_port)
{
    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if(fd < 0){
        LOGE("socket() fail.[%d]", errno);
        return -1;
    }

#if 0
    struct sockaddr_in src_sa4;
    memset(&src_sa4, 0, sizeof(src_sa4));
    src_sa4.sin_family = AF_INET;
    src_sa4.sin_port = htons(0);
    src_sa4.sin_addr.s_addr = inet_addr("127.0.0.1");

    if (bind(fd, (struct sockaddr*) &src_sa4, sizeof(src_sa4)) < 0) {
        LOGE("bind() failed: %d", errno);
        goto result_fail_with_close;
    }
#else
    if(strlen(vlan) > 0) {
        if (setsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE,
                       vlan, strlen(vlan)) < 0) {
            LOGE("setsockopt(SO_BINDTODEVICE) failed");
            return -1;
        }
    }
#endif

    struct sockaddr_in dst_sa4;
    if (inet_pton(AF_INET, remote_addr, &dst_sa4.sin_addr) > 0) {
        dst_sa4.sin_family = AF_INET;
        dst_sa4.sin_port = htons(remote_port);
        memset(&dst_sa4.sin_zero, 0, sizeof(dst_sa4.sin_zero));
    } else {
        LOGE("Set dst addr fail.");
        return -1;
    }

    if (connect(fd, (struct sockaddr*) &dst_sa4, sizeof(dst_sa4)) < 0) {
        LOGE("connect() failed: %d", errno);
        goto result_fail_with_close;
    }

    if(socket_noblock(fd)) {
        goto result_fail_with_close;
    }

    if(udp_setsockopt(fd)) {
        goto result_fail_with_close;
    }

    return fd;
result_fail_with_close:
    close(fd);
    fd = -1;
    LOGE("mbtk_sock_open() end:fail");
    return -1;
}


/*===========================================================================
FUNCTION rtp_udp_server_start

DESCRIPTION:
  Start RTP UDP server,will monitor rtp packet from MCU.

PARAMETERS:
  conf_info : RTP config informations.

RETURN VALUE:
  int : 0 for success, other for fail.

===========================================================================*/
int rtp_udp_server_start(rtp_config_t *conf_info)
{
    UNUSED(conf_info);
    if(udp_state != RTP_UDP_SER_STATE_IDEL) {
        LOGE("udp_state error : %d", udp_state);
        return -1;
    }

    udp_state = RTP_UDP_SER_STATE_STARTING;
    rtp_info.frame_size = conf_info->channel * RTP_SAMPLE_NUMBER;
    rtp_info.send.mtu = (RTP_DEFAULT_MTU / rtp_info.frame_size) * rtp_info.frame_size;
    if(conf_info->sample_rate == MBTK_AUDIO_SAMPLE_RATE_8000) {
        rtp_info.playback_size = MBTK_PCM_NB_BUF_SIZE;
        rtp_info.sample_for_ms = 8;
    } else {
        rtp_info.playback_size = MBTK_PCM_WB_BUF_SIZE;
        rtp_info.sample_for_ms = 16;
    }
    LOGD("frame_size = %d, MTU = %d", rtp_info.frame_size, rtp_info.send.mtu);

    // Open UDP server socket.
    LOGD("Start open UDP server : NULL-%d", conf_info->client_port);
    rtp_info.udp_recv_sock.fd = rtp_udp_ser_open(NULL, conf_info->client_port);
    if(rtp_info.udp_recv_sock.fd < 0) {
        LOGE("socket(udp_recv_sock) fail : errno = %d", errno);
        goto fail;
    }
    rtp_info.udp_recv_sock.read_cb = udp_read_cb;
    socket_noblock(rtp_info.udp_recv_sock.fd);
    epoll_fd_add(rtp_info.udp_recv_sock.fd);

#if 0
    rtp_info.udp_send_sock.fd = rtp_udp_cli_open(conf_info->remote_ip, conf_info->server_port);
    if(rtp_info.udp_send_sock.fd < 0) {
        LOGW("socket(udp_send_sock) fail : errno = %d", errno);
        LOGW("Can not connected to %s:%d.", conf_info->remote_ip, conf_info->server_port);
        // goto fail;
    }
    rtp_info.udp_send_sock.read_cb = NULL;
#endif

    udp_state = RTP_UDP_SER_STATE_RUNNING;

    LOGD("UDP server is running...");
    return 0;

fail:
    if(rtp_info.udp_recv_sock.fd > 0) {
        epoll_fd_del(rtp_info.udp_recv_sock.fd);
        close(rtp_info.udp_recv_sock.fd);
        rtp_info.udp_recv_sock.fd = -1;
        rtp_info.udp_recv_sock.read_cb = NULL;
    }

#if 0
    if(rtp_info.udp_send_sock.fd > 0) {
        close(rtp_info.udp_send_sock.fd);
        rtp_info.udp_send_sock.fd = -1;
        rtp_info.udp_send_sock.read_cb = NULL;
    }
#endif

    udp_state = RTP_UDP_SER_STATE_IDEL;
    return -1;
}

/*===========================================================================
FUNCTION rtp_udp_server_stop

DESCRIPTION:
  Stop RTP UDP server.

PARAMETERS:
  Non.

RETURN VALUE:
  int : 0 for success, other for fail.

===========================================================================*/
int rtp_udp_server_stop()
{
    if(udp_state != RTP_UDP_SER_STATE_RUNNING) {
        LOGE("udp_state error : %d", udp_state);
        return -1;
    }

    udp_state = RTP_UDP_SER_STATE_STOPING;
    if(rtp_info.udp_recv_sock.fd > 0) {
        epoll_fd_del(rtp_info.udp_recv_sock.fd);

        close(rtp_info.udp_recv_sock.fd);
        rtp_info.udp_recv_sock.fd = -1;
        rtp_info.udp_recv_sock.read_cb = NULL;
    }

#if 0
    if(rtp_info.udp_send_sock.fd > 0) {
        close(rtp_info.udp_send_sock.fd);
        rtp_info.udp_send_sock.fd = -1;
        rtp_info.udp_send_sock.read_cb = NULL;
    }
#endif

    udp_state = RTP_UDP_SER_STATE_IDEL;

    return 0;
}


/*===========================================================================
FUNCTION rtp_voip_server_start

DESCRIPTION:
  Start RTP voip server.will start playback/record PCM to/from voice path.

PARAMETERS:
  conf_info : RTP config informations.

RETURN VALUE:
  int : 0 for success, other for fail.

===========================================================================*/
int rtp_voip_server_start(const rtp_config_t *conf_info)
{
    UNUSED(conf_info);
    if(voip_state != RTP_VOIP_SER_STATE_IDEL) {
        LOGE("voip_state error : %d", voip_state);
        return -1;
    }

    voip_state = RTP_VOIP_SER_STATE_STARTING;
    if(mbtk_audio_pcm_init()) {
        LOGE("mbtk_audio_pcm_init() fail.");
        voip_state = RTP_VOIP_SER_STATE_IDEL;
    }

    LOGD("Start open UDP client : %s-%d", conf_info->remote_ip, conf_info->server_port);
    rtp_info.udp_send_sock.fd = rtp_udp_cli_open(conf_info->vlan, conf_info->remote_ip, conf_info->server_port);
    if(rtp_info.udp_send_sock.fd < 0) {
        LOGE("Can not connected to %s:%d [errno-%d].", conf_info->remote_ip, conf_info->server_port, errno);
        goto error;
    }
    rtp_info.udp_send_sock.read_cb = NULL;

    rtp_send_init();

    if(mbtk_audio_voice_pcm_record_start(voip_recorder_cb)) {
        LOGE("mbtk_audio_voice_pcm_record_start() fail.");
        goto error;
    }

    voip_state = RTP_VOIP_SER_STATE_RUNNING;

    pthread_attr_t thread_attr;
    pthread_attr_init(&thread_attr);
    if(pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED))
    {
        LOGE("pthread_attr_setdetachstate() fail.");
        goto error;
    }

    if (pthread_create(&voip_playback, NULL, (void *)&voip_playback_thread, NULL) < 0) {
        LOGE("%s: error creating thread_recorder!", __FUNCTION__);
        goto error;
    }

    LOGD("VOIP server is running...");
    return 0;
error:
    if(mbtk_audio_pcm_deinit()) {
        LOGE("mbtk_audio_pcm_deinit() fail.");
    }

    if(rtp_info.udp_send_sock.fd > 0) {
        close(rtp_info.udp_send_sock.fd);
        rtp_info.udp_send_sock.fd = -1;
        rtp_info.udp_send_sock.read_cb = NULL;
    }

    voip_state = RTP_VOIP_SER_STATE_IDEL;
    return -1;
}


/*===========================================================================
FUNCTION rtp_server_stop

DESCRIPTION:
  Stop RTP voip server.

PARAMETERS:
  Non.

RETURN VALUE:
  int : 0 for success, other for fail.

===========================================================================*/
int rtp_voip_server_stop()
{
    if(voip_state != RTP_VOIP_SER_STATE_RUNNING) {
        LOGE("voip_state error : %d", voip_state);
        return -1;
    }

    voip_state = RTP_VOIP_SER_STATE_STOPING;
#if 0
    if(mbtk_audio_pcm_play_stop()) {
        LOGE("mbtk_audio_pcm_play_stop() fail.");
        return -1;
    }
#else
    if (pthread_join(voip_playback, NULL)) {
        LOGE("error join voip_playback!");
        return -1;
    }
#endif

    if(mbtk_audio_pcm_recorder_stop()) {
        LOGE("mbtk_audio_pcm_recorder_stop() fail.");
        return -1;
    }

    if(mbtk_audio_pcm_deinit()) {
        LOGE("mbtk_audio_pcm_deinit() fail.");
        return -1;
    }

    if(rtp_info.recv.recv_buff) {
        mbtk_loopbuff_free(rtp_info.recv.recv_buff);
        rtp_info.recv.recv_buff = NULL;
    }

    voip_state = RTP_VOIP_SER_STATE_IDEL;

    return 0;
}