Decouple subscription handling from NOTIFY/PUBLISH body generation.

When the PJSIP pubsub framework was created, subscription handlers were required
to state what event they handled along with what body types they knew how to
generate. While this serves well when implementing a base RFC, it has problems
when trying to extend the body to support non-standard or proprietary body
elements. The code also was NOTIFY-specific, meaning that when the time comes
that we start writing code to send out PUBLISH requests with MWI or presence
bodies, we would likely find ourselves duplicating code that had previously been
written.

This changeset introduces the concept of body generators and body supplements. A
body generator is responsible for allocating a native structure for a given body
type, providing the primary body content, converting the native structure to a
string, and deallocating resources. A body supplement takes the primary body
content (the native structure, not a string) generated by the body generator and
adds nonstandard elements to the body. With these elements living in their own
module, it becomes easy to extend our support for body types and to re-use
resources when sending a PUBLISH request.

Body generators and body supplements register themselves with the pubsub core,
similar to how subscription and publish handlers had done. Now, subscription
handlers do not need to know what type of body content they generate, but they
still need to inform the pubsub core about what the default body type for a
given event package is. The pubsub core keeps track of what body generators and
body supplements have been registered. When a SUBSCRIBE arrives, the pubsub core
will check that there is a subscription handler for the event in the SUBSCRIBE,
then it will check that there is a body generator that can provide the content
specified in the Accept header(s).

Because of the nature of body generators and supplements, it means
res_pjsip_exten_state and res_pjsip_mwi have been completely gutted. They no
longer worry about body types, instead calling
ast_sip_pubsub_generate_body_content() when they need to generate a NOTIFY body.

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

Merged revisions 407016 from http://svn.asterisk.org/svn/asterisk/branches/12


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

1 files changed, 212 insertions, 56 deletions

diff --git a/res/res_pjsip_pubsub.c b/res/res_pjsip_pubsub.c
index 5bc2cb468..f10bf41fc 100644
--- a/res/res_pjsip_pubsub.c
+++ b/res/res_pjsip_pubsub.c
@@ -81,6 +81,8 @@ static struct pjsip_module pubsub_module = {
 	.on_rx_request = pubsub_on_rx_request,
 };
 
+#define MOD_DATA_BODY_GENERATOR "sub_body_generator"
+
 static const pj_str_t str_event_name = { "Event", 5 };
 
 /*! \brief Scheduler used for automatically expiring publications */
@@ -195,6 +197,8 @@ struct ast_sip_subscription {
 	pjsip_evsub *evsub;
 	/*! The underlying PJSIP dialog */
 	pjsip_dialog *dlg;
+	/*! Body generaator for NOTIFYs */
+	struct ast_sip_pubsub_body_generator *body_generator;
 	/*! Next item in the list */
 	AST_LIST_ENTRY(ast_sip_subscription) next;
 };
@@ -206,6 +210,9 @@ static const char *sip_subscription_roles_map[] = {
 
 AST_RWLIST_HEAD_STATIC(subscriptions, ast_sip_subscription);
 
+AST_RWLIST_HEAD_STATIC(body_generators, ast_sip_pubsub_body_generator);
+AST_RWLIST_HEAD_STATIC(body_supplements, ast_sip_pubsub_body_supplement);
+
 static void add_subscription(struct ast_sip_subscription *obj)
 {
 	SCOPED_LOCK(lock, &subscriptions, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
@@ -402,6 +409,8 @@ struct ast_sip_subscription *ast_sip_create_subscription(const struct ast_sip_su
 		ao2_ref(sub, -1);
 		return NULL;
 	}
+	sub->body_generator = ast_sip_mod_data_get(rdata->endpt_info.mod_data,
+			pubsub_module.id, MOD_DATA_BODY_GENERATOR);
 	sub->role = role;
 	if (role == AST_SIP_NOTIFIER) {
 		dlg = ast_sip_create_dialog_uas(endpoint, rdata);
@@ -631,31 +640,35 @@ static void sub_add_handler(struct ast_sip_subscription_handler *handler)
 	ast_module_ref(ast_module_info->self);
 }
 
-static int sub_handler_exists_for_event_name(const char *event_name)
+static struct ast_sip_subscription_handler *find_sub_handler_for_event_name(const char *event_name)
 {
 	struct ast_sip_subscription_handler *iter;
 	SCOPED_LOCK(lock, &subscription_handlers, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
 
 	AST_RWLIST_TRAVERSE(&subscription_handlers, iter, next) {
 		if (!strcmp(iter->event_name, event_name)) {
-			return 1;
+			break;
 		}
 	}
-	return 0;
+	return iter;
 }
 
 int ast_sip_register_subscription_handler(struct ast_sip_subscription_handler *handler)
 {
+	pj_str_t event;
 	pj_str_t accept[AST_SIP_MAX_ACCEPT];
+	struct ast_sip_subscription_handler *existing;
 	int i;
 
 	if (ast_strlen_zero(handler->event_name)) {
-		ast_log(LOG_ERROR, "No event package specifief for subscription handler. Cannot register\n");
+		ast_log(LOG_ERROR, "No event package specified for subscription handler. Cannot register\n");
 		return -1;
 	}
 
-	if (ast_strlen_zero(handler->accept[0])) {
-		ast_log(LOG_ERROR, "Subscription handler must supply at least one 'Accept' format\n");
+	existing = find_sub_handler_for_event_name(handler->event_name);
+	if (existing) {
+		ast_log(LOG_ERROR, "Unable to register subscription handler for event %s."
+				"A handler is already registered\n", handler->event_name);
 		return -1;
 	}
 
@@ -663,19 +676,12 @@ int ast_sip_register_subscription_handler(struct ast_sip_subscription_handler *h
 		pj_cstr(&accept[i], handler->accept[i]);
 	}
 
-	if (!sub_handler_exists_for_event_name(handler->event_name)) {
-		pj_str_t event;
-
-		pj_cstr(&event, handler->event_name);
+	pj_cstr(&event, handler->event_name);
 
-		if (!strcmp(handler->event_name, "message-summary")) {
-			pjsip_mwi_init_module(ast_sip_get_pjsip_endpoint(), pjsip_evsub_instance());
-		} else {
-			pjsip_evsub_register_pkg(&pubsub_module, &event, DEFAULT_EXPIRES, i, accept);
-		}
+	if (!strcmp(handler->event_name, "message-summary")) {
+		pjsip_mwi_init_module(ast_sip_get_pjsip_endpoint(), pjsip_evsub_instance());
 	} else {
-		pjsip_endpt_add_capability(ast_sip_get_pjsip_endpoint(), &pubsub_module, PJSIP_H_ACCEPT, NULL,
-			i, accept);
+		pjsip_evsub_register_pkg(&pubsub_module, &event, DEFAULT_EXPIRES, i, accept);
 	}
 
 	sub_add_handler(handler);
@@ -696,48 +702,52 @@ void ast_sip_unregister_subscription_handler(struct ast_sip_subscription_handler
 	AST_RWLIST_TRAVERSE_SAFE_END;
 }
 
-static struct ast_sip_subscription_handler *find_sub_handler(const char *event, char accept[AST_SIP_MAX_ACCEPT][64], size_t num_accept)
+static struct ast_sip_pubsub_body_generator *find_body_generator_type_subtype(const char *content_type,
+		const char *content_subtype)
 {
-	struct ast_sip_subscription_handler *iter;
-	int match = 0;
-	SCOPED_LOCK(lock, &subscription_handlers, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
-	AST_RWLIST_TRAVERSE(&subscription_handlers, iter, next) {
-		int i;
-		int j;
-		if (strcmp(event, iter->event_name)) {
-			ast_debug(3, "Event %s does not match %s\n", event, iter->event_name);
-			continue;
-		}
-		ast_debug(3, "Event name match: %s = %s\n", event, iter->event_name);
-		if (!num_accept && iter->handles_default_accept) {
-			/* The SUBSCRIBE contained no Accept headers, and this subscription handler
-			 * provides the default body type, so it's a match!
-			 */
+	struct ast_sip_pubsub_body_generator *iter;
+	SCOPED_LOCK(lock, &body_generators, AST_RWLIST_RDLOCK, AST_RWLIST_UNLOCK);
+
+	AST_LIST_TRAVERSE(&body_generators, iter, list) {
+		if (!strcmp(iter->type, content_type) &&
+				!strcmp(iter->subtype, content_subtype)) {
 			break;
 		}
-		for (i = 0; i < num_accept; ++i) {
-			for (j = 0; j < num_accept; ++j) {
-				if (ast_strlen_zero(iter->accept[i])) {
-					ast_debug(3, "Breaking because subscription handler has run out of 'accept' types\n");
-					break;
-				}
-				if (!strcmp(accept[j], iter->accept[i])) {
-					ast_debug(3, "Accept headers match: %s = %s\n", accept[j], iter->accept[i]);
-					match = 1;
-					break;
-				}
-				ast_debug(3, "Accept %s does not match %s\n", accept[j], iter->accept[i]);
-			}
-			if (match) {
-				break;
-			}
-		}
-		if (match) {
+	};
+
+	return iter;
+}
+
+static struct ast_sip_pubsub_body_generator *find_body_generator_accept(const char *accept)
+{
+	char *accept_copy = ast_strdupa(accept);
+	char *subtype = accept_copy;
+	char *type = strsep(&subtype, "/");
+
+	if (ast_strlen_zero(type) || ast_strlen_zero(subtype)) {
+		return NULL;
+	}
+
+	return find_body_generator_type_subtype(type, subtype);
+}
+
+static struct ast_sip_pubsub_body_generator *find_body_generator(char accept[AST_SIP_MAX_ACCEPT][64],
+		size_t num_accept)
+{
+	int i;
+	struct ast_sip_pubsub_body_generator *generator = NULL;
+
+	for (i = 0; i < num_accept; ++i) {
+		generator = find_body_generator_accept(accept[i]);
+		if (generator) {
+			ast_debug(3, "Body generator %p found for accept type %s\n", generator, accept[i]);
 			break;
+		} else {
+			ast_debug(3, "No body generator found for accept type %s\n", accept[i]);
 		}
 	}
 
-	return iter;
+	return generator;
 }
 
 static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
@@ -751,6 +761,7 @@ static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 	RAII_VAR(struct ast_sip_endpoint *, endpoint, NULL, ao2_cleanup);
 	struct ast_sip_subscription *sub;
 	size_t num_accept_headers;
+	struct ast_sip_pubsub_body_generator *generator;
 
 	endpoint = ast_pjsip_rdata_get_endpoint(rdata);
 	ast_assert(endpoint != NULL);
@@ -778,6 +789,13 @@ static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 	}
 	ast_copy_pj_str(event, &event_header->event_type, sizeof(event));
 
+	handler = find_sub_handler_for_event_name(event);
+	if (!handler) {
+		ast_log(LOG_WARNING, "No registered subscribe handler for event %s\n", event);
+		pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 489, NULL, NULL, NULL);
+		return PJ_TRUE;
+	}
+
 	accept_header = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_ACCEPT, rdata->msg_info.msg->hdr.next);
 	if (accept_header) {
 		int i;
@@ -787,15 +805,22 @@ static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 		}
 		num_accept_headers = accept_header->count;
 	} else {
-		num_accept_headers = 0;
+		/* If a SUBSCRIBE contains no Accept headers, then we must assume that
+		 * the default accept type for the event package is to be used.
+		 */
+		ast_copy_string(accept[0], handler->default_accept, sizeof(accept[0]));
+		num_accept_headers = 1;
 	}
 
-	handler = find_sub_handler(event, accept, num_accept_headers);
-	if (!handler) {
-		ast_log(LOG_WARNING, "No registered subscribe handler for event %s\n", event);
+	generator = find_body_generator(accept, num_accept_headers);
+	if (!generator) {
 		pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 489, NULL, NULL, NULL);
 		return PJ_TRUE;
 	}
+
+	ast_sip_mod_data_set(rdata->tp_info.pool, rdata->endpt_info.mod_data,
+			pubsub_module.id, MOD_DATA_BODY_GENERATOR, generator);
+
 	sub = handler->new_subscribe(endpoint, rdata);
 	if (!sub) {
 		pjsip_transaction *trans = pjsip_rdata_get_tsx(rdata);
@@ -1063,6 +1088,137 @@ pj_status_t ast_sip_publication_send_response(struct ast_sip_publication *pub, p
 	return pjsip_tsx_send_msg(tsx, tdata);
 }
 
+int ast_sip_pubsub_register_body_generator(struct ast_sip_pubsub_body_generator *generator)
+{
+	struct ast_sip_pubsub_body_generator *existing;
+	pj_str_t accept;
+	pj_size_t accept_len;
+
+	existing = find_body_generator_type_subtype(generator->type, generator->subtype);
+	if (existing) {
+		ast_log(LOG_WARNING, "Cannot register body generator of %s/%s."
+				"One is already registered.\n", generator->type, generator->subtype);
+		return -1;
+	}
+
+	AST_RWLIST_WRLOCK(&body_generators);
+	AST_LIST_INSERT_HEAD(&body_generators, generator, list);
+	AST_RWLIST_UNLOCK(&body_generators);
+
+	/* Lengths of type and subtype plus space for a slash. pj_str_t is not
+	 * null-terminated, so there is no need to allocate for the extra null
+	 * byte
+	 */
+	accept_len = strlen(generator->type) + strlen(generator->subtype) + 1;
+
+	accept.ptr = alloca(accept_len);
+	accept.slen = accept_len;
+	/* Safe use of sprintf */
+	sprintf(accept.ptr, "%s/%s", generator->type, generator->subtype);
+	pjsip_endpt_add_capability(ast_sip_get_pjsip_endpoint(), &pubsub_module,
+			PJSIP_H_ACCEPT, NULL, 1, &accept);
+
+	return 0;
+}
+
+void ast_sip_pubsub_unregister_body_generator(struct ast_sip_pubsub_body_generator *generator)
+{
+	struct ast_sip_pubsub_body_generator *iter;
+	SCOPED_LOCK(lock, &body_generators, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
+
+	AST_RWLIST_TRAVERSE_SAFE_BEGIN(&body_generators, iter, list) {
+		if (iter == generator) {
+			AST_LIST_REMOVE_CURRENT(list);
+			break;
+		}
+	}
+	AST_RWLIST_TRAVERSE_SAFE_END;
+}
+
+int ast_sip_pubsub_register_body_supplement(struct ast_sip_pubsub_body_supplement *supplement)
+{
+	AST_RWLIST_WRLOCK(&body_supplements);
+	AST_RWLIST_INSERT_TAIL(&body_supplements, supplement, list);
+	AST_RWLIST_UNLOCK(&body_supplements);
+
+	return 0;
+}
+
+void ast_sip_pubsub_unregister_body_supplement(struct ast_sip_pubsub_body_supplement *supplement)
+{
+	struct ast_sip_pubsub_body_supplement *iter;
+	SCOPED_LOCK(lock, &body_supplements, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
+
+	AST_RWLIST_TRAVERSE_SAFE_BEGIN(&body_supplements, iter, list) {
+		if (iter == supplement) {
+			AST_LIST_REMOVE_CURRENT(list);
+			break;
+		}
+	}
+	AST_RWLIST_TRAVERSE_SAFE_END;
+}
+
+const char *ast_sip_subscription_get_body_type(struct ast_sip_subscription *sub)
+{
+	return sub->body_generator->type;
+}
+
+const char *ast_sip_subscription_get_body_subtype(struct ast_sip_subscription *sub)
+{
+	return sub->body_generator->subtype;
+}
+
+int ast_sip_pubsub_generate_body_content(const char *type, const char *subtype,
+		void *data, struct ast_str **str)
+{
+	struct ast_sip_pubsub_body_supplement *supplement;
+	struct ast_sip_pubsub_body_generator *generator;
+	int res;
+	void *body;
+
+	generator = find_body_generator_type_subtype(type, subtype);
+	if (!generator) {
+		ast_log(LOG_WARNING, "Unable to find a body generator for %s/%s\n",
+				type, subtype);
+		return -1;
+	}
+
+	body = generator->allocate_body(data);
+	if (!body) {
+		ast_log(LOG_WARNING, "Unable to allocate a NOTIFY body of type %s/%s\n",
+				type, subtype);
+		return -1;
+	}
+
+	if (generator->generate_body_content(body, data)) {
+		res = -1;
+		goto end;
+	}
+
+	AST_RWLIST_RDLOCK(&body_supplements);
+	AST_RWLIST_TRAVERSE(&body_supplements, supplement, list) {
+		if (!strcmp(generator->type, supplement->type) &&
+				!strcmp(generator->subtype, supplement->subtype)) {
+			res = supplement->supplement_body(body, data);
+			if (res) {
+				break;
+			}
+		}
+	}
+	AST_RWLIST_UNLOCK(&body_supplements);
+
+	if (!res) {
+		generator->to_string(body, str);
+	}
+
+end:
+	if (generator->destroy_body) {
+		generator->destroy_body(body);
+	}
+
+	return res;
+}
+
 static pj_bool_t pubsub_on_rx_request(pjsip_rx_data *rdata)
 {
 	if (!pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, pjsip_get_subscribe_method())) {