[asterisk-commits] mmichelson: branch mmichelson/rls-notify r417793 - /team/mmichelson/rls-notif...

[SVN commits to the Asterisk project]

Author: mmichelson
Date: Wed Jul  2 19:32:07 2014
New Revision: 417793

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=417793
Log:
Add support for batched notifications


Modified:
    team/mmichelson/rls-notify/res/res_pjsip_pubsub.c

Modified: team/mmichelson/rls-notify/res/res_pjsip_pubsub.c
URL: http://svnview.digium.com/svn/asterisk/team/mmichelson/rls-notify/res/res_pjsip_pubsub.c?view=diff&rev=417793&r1=417792&r2=417793
==============================================================================
--- team/mmichelson/rls-notify/res/res_pjsip_pubsub.c (original)
+++ team/mmichelson/rls-notify/res/res_pjsip_pubsub.c Wed Jul  2 19:32:07 2014
@@ -359,6 +359,12 @@
 	pjsip_evsub *evsub;
 	/*! The underlying PJSIP dialog */
 	pjsip_dialog *dlg;
+	/*! Interval to use for batching notifications */
+	unsigned int notification_batch_interval;
+	/*! Scheduler ID for batched notification */
+	int notify_sched_id;
+	/*! Indicator if scheduled batched notification should be sent */
+	unsigned int send_scheduled_notify;
 };
 
 /*!
@@ -636,9 +642,11 @@
 	return find_body_generator(accept, num_accept_headers);
 }
 
+struct resource_tree;
+
 static struct ast_sip_subscription *create_real_subscription(const struct ast_sip_subscription_handler *handler,
 		struct ast_sip_endpoint *endpoint, pjsip_rx_data *rdata, const char *resource,
-		struct ast_sip_pubsub_body_generator *generator);
+		struct ast_sip_pubsub_body_generator *generator, struct resource_tree *tree);
 
 /*!
  * \brief A node for a resource tree.
@@ -822,6 +830,7 @@
  */
 struct resource_tree {
 	struct tree_node *root;
+	unsigned int notification_batch_interval;
 };
 
 /*!
@@ -871,7 +880,10 @@
 	}
 
 	AST_VECTOR_INIT(&visited, AST_VECTOR_SIZE(&list->items));
+
 	tree->root = tree_node_alloc(resource, &visited);
+	tree->notification_batch_interval = list->notification_batch_interval;
+
 	build_node_children(endpoint, handler, list, tree->root, &visited);
 	AST_VECTOR_FREE(&visited);
 	ao2_cleanup(list);
@@ -1052,7 +1064,7 @@
 	struct ast_sip_subscription *sub;
 
 	/* Start by creating the root subscription. It's the only real subscription. */
-	sub = create_real_subscription(handler, endpoint, rdata, resource, generator);
+	sub = create_real_subscription(handler, endpoint, rdata, resource, generator, tree);
 	if (!sub) {
 		return NULL;
 	}
@@ -1287,7 +1299,7 @@
 
 static struct ast_sip_subscription *create_real_subscription(const struct ast_sip_subscription_handler *handler,
 		struct ast_sip_endpoint *endpoint, pjsip_rx_data *rdata, const char *resource,
-		struct ast_sip_pubsub_body_generator *generator)
+		struct ast_sip_pubsub_body_generator *generator, struct resource_tree *tree)
 {
 	struct ast_sip_subscription *sub;
 	pjsip_dialog *dlg;
@@ -1323,6 +1335,7 @@
 
 	ast_sip_mod_data_set(dlg->pool, dlg->mod_data, pubsub_module.id, MOD_DATA_MSG,
 			pjsip_msg_clone(dlg->pool, rdata->msg_info.msg));
+	sub->reality.real.notification_batch_interval = tree->notification_batch_interval;
 
 	add_subscription(sub);
 	return sub;
@@ -1459,6 +1472,8 @@
 		pjsip_tx_data_dec_ref(tdata);
 		return -1;
 	}
+
+	sub->reality.real.send_scheduled_notify = 0;
 	
 	return 0;
 }
@@ -1475,6 +1490,46 @@
 	return 0;
 }
 
+static int serialized_send_notify(void *userdata)
+{
+	struct ast_sip_subscription *sub = userdata;
+
+	/* It's possible that between when the notification was scheduled
+	 * and now, that a new SUBSCRIBE arrived, requiring full state to be
+	 * sent out in an immediate NOTIFY. If that has happened, we need to
+	 * bail out here instead of sending the batched NOTIFY.
+	 */
+	if (!sub->reality.real.send_scheduled_notify) {
+		return 0;
+	}
+
+	return send_notify(sub, 0);
+}
+
+static int sched_cb(const void *data)
+{
+	/* Why the #*(@% does the scheduler give us const data?! */
+	struct ast_sip_subscription *sub = (struct ast_sip_subscription *) data;
+
+	ast_sip_push_task(sub->serializer, serialized_send_notify, sub);
+	return 0;
+}
+
+static int schedule_notification(struct ast_sip_subscription *sub)
+{
+	/* There's already a notification scheduled */
+	if (sub->reality.real.notify_sched_id > -1) {
+		return 0;
+	}
+
+	sub->reality.real.notify_sched_id = ast_sched_add(sched, sub->reality.real.notification_batch_interval, sched_cb, ao2_bump(sub));
+	if (sub->reality.real.notify_sched_id < 0) {
+		return -1;
+	}
+	sub->reality.real.send_scheduled_notify = 1;
+	return 0;
+}
+
 int ast_sip_subscription_notify(struct ast_sip_subscription *sub, void *notify_data,
 		int terminate)
 {
@@ -1490,9 +1545,11 @@
 		}
 	}
 
-	send_notify(sub, terminate);
-
-	return 0;
+	if (sub->reality.real.notification_batch_interval) {
+		return schedule_notification(sub);
+	} else {
+		return send_notify(sub, terminate);
+	}
 }
 
 void ast_sip_subscription_get_local_uri(struct ast_sip_subscription *sub, char *buf, size_t size)