res_pjsip_keepalive: Add runtime configurable keepalive module for connection-oriented transports.

This change adds a module which is configurable using the keep_alive_interval setting in the
global section that will send a CRLF keep alive to all active connection-oriented transports at
the provided interval. This is useful because it can help keep connections open through NATs.
This functionality also exists within PJSIP but can not be controlled at runtime and requires
recompiling it.

Review: https://reviewboard.asterisk.org/r/4084/


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@425825 65c4cc65-6c06-0410-ace0-fbb531ad65f3

5 files changed, 299 insertions, 0 deletions

diff --git a/configs/samples/pjsip.conf.sample b/configs/samples/pjsip.conf.sample
index 9f995e48c..830561671 100644
--- a/configs/samples/pjsip.conf.sample
+++ b/configs/samples/pjsip.conf.sample
@@ -847,6 +847,9 @@
                                                         ; int")
 ;debug=no ; Enable/Disable SIP debug logging.  Valid options include yes|no
           ; or a host address (default: "no")
+;keep_alive_interval=20 ; The interval (in seconds) at which to send keepalive
+                        ; messages on all active connection-oriented transports
+                        ; (default: "0")
 
 
 ; MODULE PROVIDING BELOW SECTION(S): res_pjsip_acl
diff --git a/include/asterisk/res_pjsip.h b/include/asterisk/res_pjsip.h
index e7c01c43b..1fe0b040e 100644
--- a/include/asterisk/res_pjsip.h
+++ b/include/asterisk/res_pjsip.h
@@ -1964,4 +1964,11 @@ char *ast_sip_get_debug(void);
 		}						\
 	} while(0)
 
+/*!
+ * \brief Retrieve the system keep alive interval setting.
+ *
+ * \retval the keep alive interval.
+ */
+unsigned int ast_sip_get_keep_alive_interval(void);
+
 #endif /* _RES_PJSIP_H */
diff --git a/res/res_pjsip.c b/res/res_pjsip.c
index 354c43d07..d93cc91e9 100644
--- a/res/res_pjsip.c
+++ b/res/res_pjsip.c
@@ -1148,6 +1148,9 @@
 				<configOption name="max_forwards" default="70">
 					<synopsis>Value used in Max-Forwards header for SIP requests.</synopsis>
 				</configOption>
+				<configOption name="keep_alive_interval" default="0">
+					<synopsis>The interval (in seconds) to send keepalives to active connection-oriented transports.</synopsis>
+				</configOption>
 				<configOption name="type">
 					<synopsis>Must be of type 'global'.</synopsis>
 				</configOption>
diff --git a/res/res_pjsip/config_global.c b/res/res_pjsip/config_global.c
index 8e2cb2a41..1779c535d 100644
--- a/res/res_pjsip/config_global.c
+++ b/res/res_pjsip/config_global.c
@@ -42,6 +42,8 @@ struct global_config {
 	);
 	/* Value to put in Max-Forwards header */
 	unsigned int max_forwards;
+	/* The interval at which to send keep alive messages to active connection-oriented transports */
+	unsigned int keep_alive_interval;
 };
 
 static void global_destructor(void *obj)
@@ -114,6 +116,21 @@ char *ast_sip_get_debug(void)
 	return res;
 }
 
+unsigned int ast_sip_get_keep_alive_interval(void)
+{
+	unsigned int interval;
+	struct global_config *cfg = get_global_cfg();
+
+	if (!cfg) {
+		return 0;
+	}
+
+	interval = cfg->keep_alive_interval;
+	ao2_ref(cfg, -1);
+
+	return interval;
+}
+
 int ast_sip_initialize_sorcery_global(void)
 {
 	struct ast_sorcery *sorcery = ast_sip_get_sorcery();
@@ -135,6 +152,8 @@ int ast_sip_initialize_sorcery_global(void)
 			OPT_STRINGFIELD_T, 0, STRFLDSET(struct global_config, default_outbound_endpoint));
 	ast_sorcery_object_field_register(sorcery, "global", "debug", "no",
 			OPT_STRINGFIELD_T, 0, STRFLDSET(struct global_config, debug));
+	ast_sorcery_object_field_register(sorcery, "global", "keep_alive_interval", "",
+			OPT_UINT_T, 0, FLDSET(struct global_config, keep_alive_interval));
 
 	return 0;
 }
diff --git a/res/res_pjsip_keepalive.c b/res/res_pjsip_keepalive.c
new file mode 100644
index 000000000..43f77346c
--- /dev/null
+++ b/res/res_pjsip_keepalive.c
@@ -0,0 +1,267 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2014, Digium, Inc.
+ *
+ * Joshua Colp <jcolp@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+/*** MODULEINFO
+	<depend>pjproject</depend>
+	<depend>res_pjsip</depend>
+	<support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+#include <pjsip.h>
+#include <pjsip_ua.h>
+
+#include "asterisk/res_pjsip.h"
+#include "asterisk/module.h"
+#include "asterisk/astobj2.h"
+
+/*! \brief Number of buckets for keepalive transports */
+#define KEEPALIVE_TRANSPORTS_BUCKETS 53
+
+/*! \brief The keep alive packet to send */
+static const pj_str_t keepalive_packet = { "\r\n\r\n", 4 };
+
+/*! \brief Global container of active transports */
+static struct ao2_container *transports;
+
+/*! \brief Thread keeping things alive */
+static pthread_t keepalive_thread = AST_PTHREADT_NULL;
+
+/*! \brief The global interval at which to send keepalives */
+static unsigned int keepalive_interval;
+
+/*! \brief Existing transport manager callback that we need to invoke */
+static pjsip_tp_state_callback tpmgr_state_callback;
+
+/*! \brief Structure for transport to be kept alive */
+struct keepalive_transport {
+	/*! \brief The underlying PJSIP transport */
+	pjsip_transport *transport;
+};
+
+/*! \brief Callback function to send keepalive */
+static int keepalive_transport_cb(void *obj, void *arg, int flags)
+{
+	struct keepalive_transport *keepalive = obj;
+	pjsip_tpselector selector = {
+		.type = PJSIP_TPSELECTOR_TRANSPORT,
+		.u.transport = keepalive->transport,
+	};
+
+	pjsip_tpmgr_send_raw(pjsip_endpt_get_tpmgr(ast_sip_get_pjsip_endpoint()),
+		keepalive->transport->key.type, &selector, NULL, keepalive_packet.ptr, keepalive_packet.slen,
+		&keepalive->transport->key.rem_addr, pj_sockaddr_get_len(&keepalive->transport->key.rem_addr),
+		NULL, NULL);
+
+	return 0;
+}
+
+/*! \brief Thread which sends keepalives to all active connection-oriented transports */
+static void *keepalive_transport_thread(void *data)
+{
+	pj_thread_desc desc;
+	pj_thread_t *thread;
+
+	if (pj_thread_register("Asterisk Keepalive Thread", desc, &thread) != PJ_SUCCESS) {
+		ast_log(LOG_ERROR, "Could not register keepalive thread with PJLIB, keepalives will not occur.\n");
+		return NULL;
+	}
+
+	/* Once loaded this module just keeps on going as it is unsafe to stop and change the underlying
+	 * callback for the transport manager.
+	 */
+	while (1) {
+		sleep(keepalive_interval);
+		ao2_callback(transports, OBJ_NODATA, keepalive_transport_cb, NULL);
+	}
+
+	return NULL;
+}
+
+/*! \brief Destructor for keepalive transport */
+static void keepalive_transport_destroy(void *obj)
+{
+	struct keepalive_transport *keepalive = obj;
+
+	pjsip_transport_dec_ref(keepalive->transport);
+}
+
+/*! \brief Callback invoked when transport changes occur */
+static void keepalive_transport_state_callback(pjsip_transport *transport, pjsip_transport_state state,
+	const pjsip_transport_state_info *info)
+{
+	/* We only care about connection-oriented transports */
+	if (transport->flag & PJSIP_TRANSPORT_RELIABLE) {
+		struct keepalive_transport *keepalive;
+
+		switch (state) {
+		case PJSIP_TP_STATE_CONNECTED:
+			keepalive = ao2_alloc(sizeof(*keepalive), keepalive_transport_destroy);
+			if (keepalive) {
+				keepalive->transport = transport;
+				pjsip_transport_add_ref(keepalive->transport);
+				ao2_link(transports, keepalive);
+				ao2_ref(keepalive, -1);
+			}
+			break;
+		case PJSIP_TP_STATE_DISCONNECTED:
+			ao2_find(transports, transport->obj_name, OBJ_SEARCH_KEY | OBJ_NODATA | OBJ_UNLINK);
+			break;
+		default:
+			break;
+		}
+	}
+
+	/* Forward to the old state callback if present */
+	if (tpmgr_state_callback) {
+		tpmgr_state_callback(transport, state, info);
+	}
+}
+
+/*! \brief Hashing function for keepalive transport */
+static int keepalive_transport_hash_fn(const void *obj, int flags)
+{
+	const struct keepalive_transport *object;
+	const char *key;
+
+	switch (flags & OBJ_SEARCH_MASK) {
+	case OBJ_SEARCH_KEY:
+		key = obj;
+		break;
+	case OBJ_SEARCH_OBJECT:
+		object = obj;
+		key = object->transport->obj_name;
+		break;
+	default:
+		/* Hash can only work on something with a full key. */
+		ast_assert(0);
+		return 0;
+	}
+	return ast_str_hash(key);
+}
+
+/*! \brief Comparison function for keepalive transport */
+static int keepalive_transport_cmp_fn(void *obj, void *arg, int flags)
+{
+	const struct keepalive_transport *object_left = obj;
+	const struct keepalive_transport *object_right = arg;
+	const char *right_key = arg;
+	int cmp;
+
+	switch (flags & OBJ_SEARCH_MASK) {
+	case OBJ_SEARCH_OBJECT:
+		right_key = object_right->transport->obj_name;
+		/* Fall through */
+	case OBJ_SEARCH_KEY:
+		cmp = strcmp(object_left->transport->obj_name, right_key);
+		break;
+	case OBJ_SEARCH_PARTIAL_KEY:
+		/*
+		 * We could also use a partial key struct containing a length
+		 * so strlen() does not get called for every comparison instead.
+		 */
+		cmp = strncmp(object_left->transport->obj_name, right_key, strlen(right_key));
+		break;
+	default:
+		/*
+		 * What arg points to is specific to this traversal callback
+		 * and has no special meaning to astobj2.
+		 */
+		cmp = 0;
+		break;
+	}
+
+	return !cmp ? CMP_MATCH | CMP_STOP : 0;
+}
+
+static void keepalive_global_loaded(const char *object_type)
+{
+	unsigned int new_interval = ast_sip_get_keep_alive_interval();
+	pjsip_tpmgr *tpmgr;
+
+	if (new_interval) {
+		keepalive_interval = new_interval;
+	} else if (keepalive_interval) {
+		ast_log(LOG_NOTICE, "Keepalive support can not be disabled once activated.\n");
+		return;
+	} else {
+		/* This will occur if no keepalive interval has been specified at initial start */
+		return;
+	}
+
+	if (keepalive_thread != AST_PTHREADT_NULL) {
+		return;
+	}
+
+	transports = ao2_container_alloc(KEEPALIVE_TRANSPORTS_BUCKETS, keepalive_transport_hash_fn,
+		keepalive_transport_cmp_fn);
+	if (!transports) {
+		ast_log(LOG_ERROR, "Could not create container for transports to perform keepalive on.\n");
+		return;
+	}
+
+	tpmgr = pjsip_endpt_get_tpmgr(ast_sip_get_pjsip_endpoint());
+	if (!tpmgr) {
+		ast_log(LOG_ERROR, "No transport manager to attach keepalive functionality to.\n");
+		ao2_ref(transports, -1);
+		return;
+	}
+
+	if (ast_pthread_create(&keepalive_thread, NULL, keepalive_transport_thread, NULL)) {
+		ast_log(LOG_ERROR, "Could not create thread for sending keepalive messages.\n");
+		ao2_ref(transports, -1);
+		return;
+	}
+
+	tpmgr_state_callback = pjsip_tpmgr_get_state_cb(tpmgr);
+	pjsip_tpmgr_set_state_cb(tpmgr, &keepalive_transport_state_callback);
+}
+
+/*! \brief Observer which is used to update our interval when the global setting changes */
+static struct ast_sorcery_observer keepalive_global_observer = {
+	.loaded = keepalive_global_loaded,
+};
+
+static int load_module(void)
+{
+	ast_sorcery_observer_add(ast_sip_get_sorcery(), "global", &keepalive_global_observer);
+	ast_sorcery_reload_object(ast_sip_get_sorcery(), "global");
+	ast_module_ref(ast_module_info->self);
+	return AST_MODULE_LOAD_SUCCESS;
+}
+
+static int unload_module(void)
+{
+	/* This will never get called */
+	return 0;
+}
+
+static int reload_module(void)
+{
+	ast_sorcery_reload_object(ast_sip_get_sorcery(), "global");
+	return 0;
+}
+
+AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_LOAD_ORDER, "PJSIP Stateful Connection Keepalive Support",
+		.support_level = AST_MODULE_SUPPORT_CORE,
+		.load = load_module,
+		.reload = reload_module,
+		.unload = unload_module,
+		.load_pri = AST_MODPRI_CHANNEL_DEPEND - 4,
+		   );