pjsip_options: Add qualify_timeout processing and eventing

This is the second follow-on to https://reviewboard.asterisk.org/r/4572/ and the
discussion at
http://lists.digium.com/pipermail/asterisk-dev/2015-March/073921.html

The basic issues are that changes in contact status don't cause events to be
emitted for the associated endpoint.  Only dynamic contact add/delete actions
update the endpoint.  Also, the qualify timeout is fixed by pjsip at 32 seconds
which is a long time.

This patch makes use of the new transaction timeout feature in r4585 and
provides the following capabilities...

1.  A new aor/contact variable 'qualify_timeout' has been added that allows the
user to specify the maximum time in milliseconds to wait for a response to an
OPTIONS message.  The default is 3000ms.  When the timer expires, the contact is
marked unavailable.

2.  Contact status changes are now propagated up to the endpoint as follows...
When any contact is 'Available', the endpoint is marked as 'Reachable'.  When
all contacts are 'Unavailable', the endpoint is marked as 'Unreachable'.  The
existing endpoint events are generated appropriately.

ASTERISK-24863 #close

Change-Id: Id0ce0528e58014da1324856ea537e7765466044a
Tested-by: Dmitriy Serov
Tested-by: George Joseph <george.joseph@fairview5.com>

3 files changed, 207 insertions, 23 deletions

diff --git a/res/res_pjsip/location.c b/res/res_pjsip/location.c
index 73ffdca0e..f784cb40f 100644
--- a/res/res_pjsip/location.c
+++ b/res/res_pjsip/location.c
@@ -188,6 +188,40 @@ struct ast_sip_contact *ast_sip_location_retrieve_contact_from_aor_list(const ch
 	return contact;
 }
 
+static int permanent_uri_sort_fn(const void *obj_left, const void *obj_right, int flags);
+static int cli_contact_populate_container(void *obj, void *arg, int flags);
+
+static int gather_contacts_for_aor(void *obj, void *arg, int flags)
+{
+	struct ao2_container *aor_contacts;
+	struct ast_sip_aor *aor = obj;
+	struct ao2_container *container = arg;
+
+	aor_contacts = ast_sip_location_retrieve_aor_contacts(aor);
+	if (!aor_contacts) {
+		return 0;
+	}
+	ao2_callback(aor_contacts, OBJ_MULTIPLE | OBJ_NODATA, cli_contact_populate_container,
+		container);
+	ao2_ref(aor_contacts, -1);
+	return CMP_MATCH;
+}
+
+struct ao2_container *ast_sip_location_retrieve_contacts_from_aor_list(const char *aor_list)
+{
+	struct ao2_container *contacts;
+
+	contacts = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK,
+		AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT, permanent_uri_sort_fn, NULL);
+	if (!contacts) {
+		return NULL;
+	}
+
+	ast_sip_for_each_aor(aor_list, gather_contacts_for_aor, contacts);
+
+	return contacts;
+}
+
 struct ast_sip_contact *ast_sip_location_retrieve_contact(const char *contact_name)
 {
 	return ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "contact", contact_name);
@@ -208,6 +242,7 @@ int ast_sip_location_add_contact(struct ast_sip_aor *aor, const char *uri,
 	ast_string_field_set(contact, uri, uri);
 	contact->expiration_time = expiration_time;
 	contact->qualify_frequency = aor->qualify_frequency;
+	contact->qualify_timeout = aor->qualify_timeout;
 	contact->authenticate_qualify = aor->authenticate_qualify;
 	if (path_info && aor->support_path) {
 		ast_string_field_set(contact, path, path_info);
@@ -853,7 +888,8 @@ int ast_sip_initialize_sorcery_location(void)
 	ast_sorcery_object_field_register(sorcery, "contact", "path", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, path));
 	ast_sorcery_object_field_register_custom(sorcery, "contact", "expiration_time", "", expiration_str2struct, expiration_struct2str, NULL, 0, 0);
 	ast_sorcery_object_field_register(sorcery, "contact", "qualify_frequency", 0, OPT_UINT_T,
-					  PARSE_IN_RANGE, FLDSET(struct ast_sip_contact, qualify_frequency), 0, 86400);
+		PARSE_IN_RANGE, FLDSET(struct ast_sip_contact, qualify_frequency), 0, 86400);
+	ast_sorcery_object_field_register(sorcery, "contact", "qualify_timeout", "3.0", OPT_DOUBLE_T, 0, FLDSET(struct ast_sip_contact, qualify_timeout));
 	ast_sorcery_object_field_register(sorcery, "contact", "outbound_proxy", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, outbound_proxy));
 	ast_sorcery_object_field_register(sorcery, "contact", "user_agent", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, user_agent));
 
@@ -862,6 +898,7 @@ int ast_sip_initialize_sorcery_location(void)
 	ast_sorcery_object_field_register(sorcery, "aor", "maximum_expiration", "7200", OPT_UINT_T, 0, FLDSET(struct ast_sip_aor, maximum_expiration));
 	ast_sorcery_object_field_register(sorcery, "aor", "default_expiration", "3600", OPT_UINT_T, 0, FLDSET(struct ast_sip_aor, default_expiration));
 	ast_sorcery_object_field_register(sorcery, "aor", "qualify_frequency", 0, OPT_UINT_T, PARSE_IN_RANGE, FLDSET(struct ast_sip_aor, qualify_frequency), 0, 86400);
+	ast_sorcery_object_field_register(sorcery, "aor", "qualify_timeout", "3.0", OPT_DOUBLE_T, 0, FLDSET(struct ast_sip_aor, qualify_timeout));
 	ast_sorcery_object_field_register(sorcery, "aor", "authenticate_qualify", "no", OPT_BOOL_T, 1, FLDSET(struct ast_sip_aor, authenticate_qualify));
 	ast_sorcery_object_field_register(sorcery, "aor", "max_contacts", "0", OPT_UINT_T, 0, FLDSET(struct ast_sip_aor, max_contacts));
 	ast_sorcery_object_field_register(sorcery, "aor", "remove_existing", "no", OPT_BOOL_T, 1, FLDSET(struct ast_sip_aor, remove_existing));
diff --git a/res/res_pjsip/pjsip_configuration.c b/res/res_pjsip/pjsip_configuration.c
index 0eecb5e0a..ab0d08449 100644
--- a/res/res_pjsip/pjsip_configuration.c
+++ b/res/res_pjsip/pjsip_configuration.c
@@ -19,6 +19,7 @@
 #include "asterisk/utils.h"
 #include "asterisk/sorcery.h"
 #include "asterisk/callerid.h"
+#include "asterisk/test.h"
 
 /*! \brief Number of buckets for persistent endpoint information */
 #define PERSISTENT_BUCKETS 53
@@ -59,31 +60,66 @@ static int persistent_endpoint_cmp(void *obj, void *arg, int flags)
 static int persistent_endpoint_update_state(void *obj, void *arg, int flags)
 {
 	struct sip_persistent_endpoint *persistent = obj;
+	struct ast_endpoint *endpoint = persistent->endpoint;
 	char *aor = arg;
-	RAII_VAR(struct ast_sip_contact *, contact, NULL, ao2_cleanup);
-	RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
+	struct ao2_container *contacts;
+	struct ast_json *blob;
+	struct ao2_iterator i;
+	struct ast_sip_contact *contact;
+	enum ast_endpoint_state state = AST_ENDPOINT_OFFLINE;
 
 	if (!ast_strlen_zero(aor) && !strstr(persistent->aors, aor)) {
 		return 0;
 	}
 
-	if ((contact = ast_sip_location_retrieve_contact_from_aor_list(persistent->aors))) {
-		ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_ONLINE);
+	/* Find all the contacts for this endpoint.  If ANY are available,
+	 * mark the endpoint as ONLINE.
+	 */
+	contacts = ast_sip_location_retrieve_contacts_from_aor_list(persistent->aors);
+	if (contacts) {
+		i = ao2_iterator_init(contacts, 0);
+		while ((contact = ao2_iterator_next(&i))
+			&& state == AST_ENDPOINT_OFFLINE) {
+			struct ast_sip_contact_status *contact_status;
+			const char *contact_id = ast_sorcery_object_get_id(contact);
+
+			contact_status = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(),
+				CONTACT_STATUS, contact_id);
+
+			if (contact_status && contact_status->status == AVAILABLE) {
+				state = AST_ENDPOINT_ONLINE;
+			}
+			ao2_cleanup(contact_status);
+			ao2_ref(contact, -1);
+		}
+		ao2_iterator_destroy(&i);
+		ao2_ref(contacts, -1);
+	}
+
+	/* If there was no state change, don't publish anything. */
+	if (ast_endpoint_get_state(endpoint) == state) {
+		return 0;
+	}
+
+	if (state == AST_ENDPOINT_ONLINE) {
+		ast_endpoint_set_state(endpoint, AST_ENDPOINT_ONLINE);
 		blob = ast_json_pack("{s: s}", "peer_status", "Reachable");
+		ast_verb(1, "Endpoint %s is now Reachable\n", ast_endpoint_get_resource(endpoint));
 	} else {
-		ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_OFFLINE);
+		ast_endpoint_set_state(endpoint, AST_ENDPOINT_OFFLINE);
 		blob = ast_json_pack("{s: s}", "peer_status", "Unreachable");
+		ast_verb(1, "Endpoint %s is now Unreachable\n", ast_endpoint_get_resource(endpoint));
 	}
 
-	ast_endpoint_blob_publish(persistent->endpoint, ast_endpoint_state_type(), blob);
-
-	ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "PJSIP/%s", ast_endpoint_get_resource(persistent->endpoint));
+	ast_endpoint_blob_publish(endpoint, ast_endpoint_state_type(), blob);
+	ast_json_unref(blob);
+	ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "PJSIP/%s", ast_endpoint_get_resource(endpoint));
 
 	return 0;
 }
 
 /*! \brief Function called when stuff relating to a contact happens (created/deleted) */
-static void persistent_endpoint_contact_observer(const void *object)
+static void persistent_endpoint_contact_created_observer(const void *object)
 {
 	char *id = ast_strdupa(ast_sorcery_object_get_id(object)), *aor = NULL;
 
@@ -92,12 +128,74 @@ static void persistent_endpoint_contact_observer(const void *object)
 	ao2_callback(persistent_endpoints, OBJ_NODATA, persistent_endpoint_update_state, aor);
 }
 
+/*! \brief Function called when stuff relating to a contact happens (created/deleted) */
+static void persistent_endpoint_contact_deleted_observer(const void *object)
+{
+	char *id = ast_strdupa(ast_sorcery_object_get_id(object));
+	char *aor = NULL;
+	char *contact = NULL;
+
+	aor = id;
+	/* Dynamic contacts are delimited with ";@" and static ones with "@@" */
+	if ((contact = strstr(id, ";@")) || (contact = strstr(id, "@@"))) {
+		*contact = '\0';
+		contact += 2;
+	} else {
+		contact = id;
+	}
+
+	ast_verb(1, "Contact %s/%s is now Unavailable\n", aor, contact);
+
+	ao2_callback(persistent_endpoints, OBJ_NODATA, persistent_endpoint_update_state, aor);
+}
+
 /*! \brief Observer for contacts so state can be updated on respective endpoints */
 static const struct ast_sorcery_observer state_contact_observer = {
-	.created = persistent_endpoint_contact_observer,
-	.deleted = persistent_endpoint_contact_observer,
+	.created = persistent_endpoint_contact_created_observer,
+	.deleted = persistent_endpoint_contact_deleted_observer,
 };
 
+/*! \brief Function called when stuff relating to a contact status happens (updated) */
+static void persistent_endpoint_contact_status_observer(const void *object)
+{
+	const struct ast_sip_contact_status *contact_status = object;
+	char *id = ast_strdupa(ast_sorcery_object_get_id(object));
+	char *aor = NULL;
+	char *contact = NULL;
+
+	/* If rtt_start is set (this is the outgoing OPTIONS) or
+	 * there's no status change, ignore.
+	 */
+	if (contact_status->rtt_start.tv_sec > 0
+		|| contact_status->status == contact_status->last_status) {
+		return;
+	}
+
+	aor = id;
+	/* Dynamic contacts are delimited with ";@" and static ones with "@@" */
+	if ((contact = strstr(id, ";@")) || (contact = strstr(id, "@@"))) {
+		*contact = '\0';
+		contact += 2;
+	} else {
+		contact = id;
+	}
+
+	ast_test_suite_event_notify("AOR_CONTACT_UPDATE",
+		"Contact: %s\r\n"
+			"Status: %s",
+		ast_sorcery_object_get_id(contact_status),
+		(contact_status->status == AVAILABLE ? "Available" : "Unavailable"));
+
+	ast_verb(1, "Contact %s/%s is now %s\n", aor, contact,
+		contact_status->status == AVAILABLE ? "Available" : "Unavailable");
+
+	ao2_callback(persistent_endpoints, OBJ_NODATA, persistent_endpoint_update_state, aor);
+}
+
+/*! \brief Observer for contacts so state can be updated on respective endpoints */
+static const struct ast_sorcery_observer state_contact_status_observer = {
+	.updated = persistent_endpoint_contact_status_observer,
+};
 
 static int dtmf_handler(const struct aco_option *opt, struct ast_variable *var, void *obj)
 {
@@ -1796,6 +1894,7 @@ int ast_res_pjsip_initialize_configuration(const struct ast_module_info *ast_mod
 	}
 
 	ast_sorcery_observer_add(sip_sorcery, "contact", &state_contact_observer);
+	ast_sorcery_observer_add(sip_sorcery, CONTACT_STATUS, &state_contact_status_observer);
 
 	if (ast_sip_initialize_sorcery_domain_alias()) {
 		ast_log(LOG_ERROR, "Failed to register SIP domain aliases support with sorcery\n");
@@ -1852,6 +1951,8 @@ int ast_res_pjsip_initialize_configuration(const struct ast_module_info *ast_mod
 
 void ast_res_pjsip_destroy_configuration(void)
 {
+	ast_sorcery_observer_remove(sip_sorcery, CONTACT_STATUS, &state_contact_status_observer);
+	ast_sorcery_observer_remove(sip_sorcery, "contact", &state_contact_observer);
 	ast_sip_destroy_sorcery_global();
 	ast_sip_destroy_sorcery_location();
 	ast_sip_destroy_sorcery_auth();
diff --git a/res/res_pjsip/pjsip_options.c b/res/res_pjsip/pjsip_options.c
index 9794827b5..dc5a70175 100644
--- a/res/res_pjsip/pjsip_options.c
+++ b/res/res_pjsip/pjsip_options.c
@@ -28,6 +28,7 @@
 #include "asterisk/astobj2.h"
 #include "asterisk/cli.h"
 #include "asterisk/time.h"
+#include "asterisk/test.h"
 #include "include/res_pjsip_private.h"
 
 #define DEFAULT_LANGUAGE "en"
@@ -110,18 +111,20 @@ static void update_contact_status(const struct ast_sip_contact *contact,
 
 	status = find_or_create_contact_status(contact);
 	if (!status) {
+		ast_log(LOG_ERROR, "Unable to find ast_sip_contact_status for contact %s\n",
+			contact->uri);
 		return;
 	}
 
 	update = ast_sorcery_alloc(ast_sip_get_sorcery(), CONTACT_STATUS,
 		ast_sorcery_object_get_id(status));
 	if (!update) {
-		ast_log(LOG_ERROR, "Unable to create update ast_sip_contact_status for contact %s\n",
+		ast_log(LOG_ERROR, "Unable to allocate ast_sip_contact_status for contact %s\n",
 			contact->uri);
-		ao2_ref(status, -1);
 		return;
 	}
 
+	update->last_status = status->status;
 	update->status = value;
 
 	/* if the contact is available calculate the rtt as
@@ -131,13 +134,21 @@ static void update_contact_status(const struct ast_sip_contact *contact,
 
 	update->rtt_start = ast_tv(0, 0);
 
+	ast_test_suite_event_notify("AOR_CONTACT_QUALIFY_RESULT",
+		"Contact: %s\r\n"
+			"Status: %s\r\n"
+			"RTT: %ld",
+		ast_sorcery_object_get_id(update),
+		(update->status == AVAILABLE ? "Available" : "Unavailable"),
+		update->rtt);
+
 	if (ast_sorcery_update(ast_sip_get_sorcery(), update)) {
 		ast_log(LOG_ERROR, "Unable to update ast_sip_contact_status for contact %s\n",
 			contact->uri);
 	}
 
-	ao2_ref(update, -1);
 	ao2_ref(status, -1);
+	ao2_ref(update, -1);
 }
 
 /*!
@@ -152,18 +163,22 @@ static void init_start_time(const struct ast_sip_contact *contact)
 
 	status = find_or_create_contact_status(contact);
 	if (!status) {
+		ast_log(LOG_ERROR, "Unable to find ast_sip_contact_status for contact %s\n",
+			contact->uri);
 		return;
 	}
 
 	update = ast_sorcery_alloc(ast_sip_get_sorcery(), CONTACT_STATUS,
 		ast_sorcery_object_get_id(status));
 	if (!update) {
-		ast_log(LOG_ERROR, "Unable to create update ast_sip_contact_status for contact %s\n",
+		ast_log(LOG_ERROR, "Unable to copy ast_sip_contact_status for contact %s\n",
 			contact->uri);
-		ao2_ref(status, -1);
 		return;
 	}
 
+	update->status = status->status;
+	update->last_status = status->last_status;
+	update->rtt = status->rtt;
 	update->rtt_start = ast_tvnow();
 
 	if (ast_sorcery_update(ast_sip_get_sorcery(), update)) {
@@ -171,8 +186,8 @@ static void init_start_time(const struct ast_sip_contact *contact)
 			contact->uri);
 	}
 
-	ao2_ref(update, -1);
 	ao2_ref(status, -1);
+	ao2_ref(update, -1);
 }
 
 /*!
@@ -320,7 +335,7 @@ static int qualify_contact(struct ast_sip_endpoint *endpoint, struct ast_sip_con
 	init_start_time(contact);
 
 	ao2_ref(contact, +1);
-	if (ast_sip_send_request(tdata, NULL, endpoint_local, contact, qualify_contact_cb)
+	if (ast_sip_send_out_of_dialog_request(tdata, endpoint_local, (int)(contact->qualify_timeout * 1000), contact, qualify_contact_cb)
 		!= PJ_SUCCESS) {
 		ast_log(LOG_ERROR, "Unable to send request to qualify contact %s\n",
 			contact->uri);
@@ -923,6 +938,32 @@ static int sched_qualifies_cmp_fn(void *obj, void *arg, int flags)
 	return CMP_MATCH;
 }
 
+static int rtt_start_handler(const struct aco_option *opt,
+	struct ast_variable *var, void *obj)
+{
+	struct ast_sip_contact_status *status = obj;
+	long int sec, usec;
+
+	if (sscanf(var->value, "%ld.%06ld", &sec, &usec) != 2) {
+		return -1;
+	}
+
+	status->rtt_start = ast_tv(sec, usec);
+
+	return 0;
+}
+
+static int rtt_start_to_str(const void *obj, const intptr_t *args, char **buf)
+{
+	const struct ast_sip_contact_status *status = obj;
+
+	if (ast_asprintf(buf, "%ld.%06ld", status->rtt_start.tv_sec, status->rtt_start.tv_usec) == -1) {
+		return -1;
+	}
+
+	return 0;
+}
+
 int ast_sip_initialize_sorcery_qualify(void)
 {
 	struct ast_sorcery *sorcery = ast_sip_get_sorcery();
@@ -936,10 +977,14 @@ int ast_sip_initialize_sorcery_qualify(void)
 		return -1;
 	}
 
-	ast_sorcery_object_field_register_nodoc(sorcery, CONTACT_STATUS, "rtt", "0", OPT_UINT_T,
-					  1, FLDSET(struct ast_sip_contact_status, status));
-	ast_sorcery_object_field_register_nodoc(sorcery, CONTACT_STATUS, "rtt", "0", OPT_UINT_T,
-					  1, FLDSET(struct ast_sip_contact_status, rtt));
+	ast_sorcery_object_field_register_nodoc(sorcery, CONTACT_STATUS, "last_status",
+		"0", OPT_UINT_T, 1, FLDSET(struct ast_sip_contact_status, last_status));
+	ast_sorcery_object_field_register_nodoc(sorcery, CONTACT_STATUS, "status",
+		"0", OPT_UINT_T, 1, FLDSET(struct ast_sip_contact_status, status));
+	ast_sorcery_object_field_register_custom_nodoc(sorcery, CONTACT_STATUS, "rtt_start",
+		"0.0", rtt_start_handler, rtt_start_to_str, NULL, 0, 0);
+	ast_sorcery_object_field_register_nodoc(sorcery, CONTACT_STATUS, "rtt",
+		"0", OPT_UINT_T, 1, FLDSET(struct ast_sip_contact_status, rtt));
 
 	return 0;
 }
@@ -951,6 +996,7 @@ static int qualify_and_schedule_cb(void *obj, void *arg, int flags)
 	int initial_interval;
 
 	contact->qualify_frequency = aor->qualify_frequency;
+	contact->qualify_timeout = aor->qualify_timeout;
 	contact->authenticate_qualify = aor->authenticate_qualify;
 
 	/* Delay initial qualification by a random fraction of the specified interval */