myvtm/Module/mod_sipphone/rvcturn.cpp

// rvcturn.cpp : 定义 DLL 应用程序的导出函数。
//

#include "stdafx.h"
#include "rvcturn.h"

static struct global
{
	pj_caching_pool	 cp;
	pj_pool_t		*pool;
	pj_stun_config	 stun_config;
	pj_thread_t		*thread;
	pj_bool_t		 quit;
	rvc_peer_t		 peer[RVC_TURN_PORT_NUM];
} g_param;


static void rvc_stun_callback(rvc_turn_callback_t* pcallback, int* ioption, void* pstuninfo, void* pdata)
{
	if (NULL != pcallback) {
		if (NULL != pcallback->status_callback) {
			pcallback->status_callback(pcallback->user_data, ioption, pstuninfo, pdata);
		}
	}
}


static void rvc_log(rvc_turn_callback_t* pcallback, const char* fmt, ...)
{
	if (NULL != pcallback) {
		if (NULL != pcallback->debug) {
			va_list arg;
			va_start(arg, fmt);
			pcallback->debug(pcallback->user_data, fmt, arg);
			va_end(arg);
		}
	}
}


static void rvcturn_log(rvc_turn_callback_t* pCallback, const char *title, pj_status_t status)
{
	char errmsg[PJ_ERR_MSG_SIZE] = {0};
	pj_strerror(status, errmsg, sizeof(errmsg));
	rvc_log(pCallback, "%s: %s", title, errmsg);
}

static int worker_thread(void *usr_data)
{
	rvc_turn_callback_t* pcallback = (rvc_turn_callback_t*)usr_data;

	while (!g_param.quit) {
		const pj_time_val delay = { 0, 10 };

		/* Poll ioqueue for the TURN client */
		pj_ioqueue_poll(g_param.stun_config.ioqueue, &delay);

		/* Poll the timer heap */
		pj_timer_heap_poll(g_param.stun_config.timer_heap, NULL);
	}
	
	rvc_log(pcallback, "%s", "worker_thread exit!");

	return 0;
}


static pj_bool_t stun_sock_on_status(pj_stun_sock *stun_sock,
	pj_stun_sock_op op,
	pj_status_t status)
{
	rvc_peer_t *peer = (rvc_peer_t*)pj_stun_sock_get_user_data(stun_sock);

	if (status == PJ_SUCCESS) {
		rvc_log(&peer->stun_status_cb, "peer%d: %s(%d) success", peer - g_param.peer, pj_stun_sock_op_name(op), op);
	}
	else {
		char errmsg[PJ_ERR_MSG_SIZE] = { 0 };
		pj_strerror(status, errmsg, sizeof(errmsg));
		rvc_log(&peer->stun_status_cb, "peer%d: %s error: %s", peer - g_param.peer, pj_stun_sock_op_name(op), errmsg);
		return PJ_FALSE;
	}

	if (op == PJ_STUN_SOCK_BINDING_OP || op == PJ_STUN_SOCK_KEEP_ALIVE_OP) {
		pj_stun_sock_info info;
		int cmp;

		pj_stun_sock_get_info(stun_sock, &info);
		cmp = pj_sockaddr_cmp(&info.mapped_addr, &peer->mapped_addr);

		if (cmp) {
			rvc_stun_info_t tinfo = { 0 };
			pj_sockaddr_cp(&peer->mapped_addr, &info.mapped_addr);
			pj_sockaddr_print(&peer->mapped_addr, tinfo.strmappedip, sizeof(tinfo.strmappedip), 2);
			tinfo.imappedport = pj_sockaddr_get_port(&peer->mapped_addr);
			
			pj_sockaddr_cp(&peer->bound_addr, &info.bound_addr);
			pj_sockaddr_print(&peer->bound_addr, tinfo.strboundip, sizeof(tinfo.strboundip), 2);
			tinfo.iboundport = pj_sockaddr_get_port(&peer->bound_addr);
			
			pj_sockaddr_cp(&peer->srv_addr, &info.srv_addr);
			pj_sockaddr_print(&peer->srv_addr, tinfo.strservip, sizeof(tinfo.strservip), 2);
			tinfo.iservport = pj_sockaddr_get_port(&peer->srv_addr);

			int ioption = op;
			rvc_stun_callback(&peer->stun_status_cb, &ioption, &tinfo, NULL);
			rvc_log(&peer->stun_status_cb, "peer%d: local bound address is %s:%d", peer - g_param.peer, tinfo.strboundip, tinfo.iboundport);
			rvc_log(&peer->stun_status_cb, "peer%d: STUN server address is %s:%d", peer - g_param.peer, tinfo.strservip, tinfo.iservport);
			rvc_log(&peer->stun_status_cb, "peer%d: STUN mapped address is %s:%d", peer - g_param.peer, tinfo.strmappedip, tinfo.imappedport);
		}
	}

	return PJ_TRUE;
}


static pj_bool_t stun_sock_on_rx_data(pj_stun_sock *stun_sock,
	void *pkt,
	unsigned pkt_len,
	const pj_sockaddr_t *src_addr,
	unsigned addr_len)
{
	rvc_peer_t *peer = (rvc_peer_t*)pj_stun_sock_get_user_data(stun_sock);
	char straddr[PJ_INET6_ADDRSTRLEN + 10];

	((char*)pkt)[pkt_len] = '\0';

	pj_sockaddr_print(src_addr, straddr, sizeof(straddr), 3);
	rvc_log(&peer->stun_status_cb, "peer%d: received %d bytes data from %s: %s", peer - g_param.peer, pkt_len, straddr, (char*)pkt);

	return PJ_TRUE;
}


#define RVC_CHECK(expr)	status=expr; \
			if (status!=PJ_SUCCESS) { \
			    rvcturn_log(pcallback, #expr, status); \
			    return status; \
			}


static int init(rvc_turn_param_t* pparam, rvc_turn_callback_t* pcallback)
{
	pj_status_t status = PJ_EINVAL;
	if (NULL == pparam || NULL == pcallback) {
		return status;
	}

	RVC_CHECK(pj_init());
	RVC_CHECK(pjlib_util_init());
	RVC_CHECK(pjnath_init());

	/* Check that server is specified */
	if (!pparam->pstrserv && !pparam->pdefaultserv) {
		rvc_log(pcallback, "%s", "Error: server must be specified.");
		return PJ_EINVAL;
	}

	pj_caching_pool_init(&g_param.cp, &pj_pool_factory_default_policy, 0);

	g_param.quit = 0;
	g_param.pool = pj_pool_create(&g_param.cp.factory, "main", 1000, 1000, NULL);

	/* Init global STUN config */
	pj_stun_config_init(&g_param.stun_config, &g_param.cp.factory, 0, NULL, NULL);

	/* Create global timer heap */
	RVC_CHECK(pj_timer_heap_create(g_param.pool, 1000, &g_param.stun_config.timer_heap));

	/* Create global ioqueue */
	RVC_CHECK(pj_ioqueue_create(g_param.pool, 16, &g_param.stun_config.ioqueue));

	/*
	* Create peers
	*/
	for (int i = 0; i<(int)PJ_ARRAY_SIZE(g_param.peer); ++i) {
		pj_stun_sock_cb stun_sock_cb;
		char name[] = "rvcpeer0";
		pj_uint16_t port;
		pj_stun_sock_cfg ss_cfg;
		pj_str_t server;

		pj_bzero(&stun_sock_cb, sizeof(stun_sock_cb));
		stun_sock_cb.on_rx_data = &stun_sock_on_rx_data;
		stun_sock_cb.on_status = &stun_sock_on_status;

		//g_param.peer[i].mapped_addr.addr.sa_family = pj_AF_INET();
		pj_sockaddr_init(pj_AF_INET(), &g_param.peer[i].mapped_addr, NULL, 0);
		memcpy(&g_param.peer[i].stun_status_cb, pcallback, sizeof(rvc_turn_callback_t));

		pj_stun_sock_cfg_default(&ss_cfg);
		pj_sockaddr_init(pj_AF_INET(), &ss_cfg.bound_addr, NULL, pparam->ilocalport[i]);

		if (pparam->bkeepalive) {
			/* make reading the log easier */
			ss_cfg.ka_interval = pparam->ikainterval;
		}

		name[strlen(name) - 1] = '0' + i;
		status = pj_stun_sock_create(&g_param.stun_config, name, pj_AF_INET(),
			&stun_sock_cb, &ss_cfg,
			&g_param.peer[i], &g_param.peer[i].stun_sock);
		if (status != PJ_SUCCESS) {
			rvcturn_log(pcallback, "pj_stun_sock_create()", status);
			return status;
		}

		if (pparam->pstrserv) {
			server = pj_str(pparam->pstrserv);
			port = (pj_uint16_t)(pparam->pserport ? atoi(pparam->pserport) : PJ_STUN_PORT);
		}
		else {
			server = pj_str(pparam->pdefaultserv);
			port = PJ_STUN_PORT;
		}
		status = pj_stun_sock_start(g_param.peer[i].stun_sock, &server, port, NULL);
		rvcturn_log(pcallback, "pj_stun_sock_start()", status);
		if (status != PJ_SUCCESS) {
			return status;
		}
	}

	/* Start the worker thread */
	RVC_CHECK(pj_thread_create(g_param.pool, "stun", &worker_thread, pcallback, 0, 0, &g_param.thread));

	return PJ_SUCCESS;
}


static int client_shutdown()
{
	if (g_param.thread) {
		g_param.quit = 1;
		pj_thread_join(g_param.thread);
		pj_thread_destroy(g_param.thread);
		g_param.thread = NULL;
	}

	rvc_log(&g_param.peer[0].stun_status_cb, "%s", "pj_thread_join success!");

	for (unsigned i = 0; i<PJ_ARRAY_SIZE(g_param.peer); ++i) {
		if (g_param.peer[i].stun_sock) {
			pj_stun_sock_destroy(g_param.peer[i].stun_sock);
			g_param.peer[i].stun_sock = NULL;
		}
	}

	rvc_log(&g_param.peer[0].stun_status_cb, "%s", "pj_stun_sock_destroy success!");

	if (g_param.stun_config.timer_heap) {
		pj_timer_heap_destroy(g_param.stun_config.timer_heap);
		g_param.stun_config.timer_heap = NULL;
	}

	rvc_log(&g_param.peer[0].stun_status_cb, "%s", "pj_timer_heap_destroy success!");

	if (g_param.stun_config.ioqueue) {
		pj_ioqueue_destroy(g_param.stun_config.ioqueue);
		g_param.stun_config.ioqueue = NULL;
	}

	rvc_log(&g_param.peer[0].stun_status_cb, "%s", "pj_ioqueue_destroy success!");

	if (g_param.pool) {
		pj_pool_release(g_param.pool);
		g_param.pool = NULL;
	}

	pj_pool_factory_dump(&g_param.cp.factory, PJ_TRUE);
	pj_caching_pool_destroy(&g_param.cp);

	return PJ_SUCCESS;
}



int start_turn_connect(rvc_turn_param_t* pparam, rvc_turn_callback_t* pcallback)
{
	int iret = -1;

	iret = init(pparam, pcallback);

	return iret;
}


int stop_turn_connect()
{
	int iret = -1;

	iret = client_shutdown();

	return iret;
}