[asterisk-commits] marquis: branch marquis/pubsub-distributed-events r213622 - in /team/marquis/...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Fri Aug 21 12:35:22 CDT 2009


Author: marquis
Date: Fri Aug 21 12:35:17 2009
New Revision: 213622

URL: http://svn.asterisk.org/svn-view/asterisk?view=rev&rev=213622
Log:
Initial commit of PubSub-based distributed events.  Mostly working for device states, with some error conditions not totally taken care of yet.

Modified:
    team/marquis/pubsub-distributed-events/include/asterisk/jabber.h
    team/marquis/pubsub-distributed-events/res/res_jabber.c

Modified: team/marquis/pubsub-distributed-events/include/asterisk/jabber.h
URL: http://svn.asterisk.org/svn-view/asterisk/team/marquis/pubsub-distributed-events/include/asterisk/jabber.h?view=diff&rev=213622&r1=213621&r2=213622
==============================================================================
--- team/marquis/pubsub-distributed-events/include/asterisk/jabber.h (original)
+++ team/marquis/pubsub-distributed-events/include/asterisk/jabber.h Fri Aug 21 12:35:17 2009
@@ -82,7 +82,8 @@
 
 enum {
 	AJI_AUTOPRUNE = (1 << 0),
-	AJI_AUTOREGISTER = (1 << 1)
+	AJI_AUTOREGISTER = (1 << 1),
+	AJI_XEP0248 = ( 1 << 2)
 };
 
 enum aji_btype {
@@ -142,6 +143,7 @@
 	char password[160];
 	char user[AJI_MAX_JIDLEN];
 	char serverhost[AJI_MAX_RESJIDLEN];
+	char pubsub_node[AJI_MAX_RESJIDLEN];
 	char statusmessage[256];
 	char name_space[256];
 	char sid[10]; /* Session ID */
@@ -167,6 +169,7 @@
 	int timeout;
 	int message_timeout;
 	int authorized;
+	int distribute_events;
 	struct ast_flags flags;
 	int component; /* 0 client,  1 component */
 	struct aji_buddy_container buddies;

Modified: team/marquis/pubsub-distributed-events/res/res_jabber.c
URL: http://svn.asterisk.org/svn-view/asterisk/team/marquis/pubsub-distributed-events/res/res_jabber.c?view=diff&rev=213622&r1=213621&r2=213622
==============================================================================
--- team/marquis/pubsub-distributed-events/res/res_jabber.c (original)
+++ team/marquis/pubsub-distributed-events/res/res_jabber.c Fri Aug 21 12:35:17 2009
@@ -58,6 +58,8 @@
 #include "asterisk/astobj.h"
 #include "asterisk/astdb.h"
 #include "asterisk/manager.h"
+#include "asterisk/event.h"
+#include "asterisk/devicestate.h"
 
 /*** DOCUMENTATION
 	<application name="JabberSend" language="en_US">
@@ -241,7 +243,24 @@
 static int aji_register_query_handler(void *data, ikspak *pak);
 static int aji_register_approve_handler(void *data, ikspak *pak);
 static int aji_reconnect(struct aji_client *client);
+static char *aji_cli_create_collection(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
+static char *aji_cli_list_pubsub_items(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
 static iks *jabber_make_auth(iksid * id, const char *pass, const char *sid);
+static int aji_receive_item_list(void *data, ikspak* pak);
+static void aji_init_event_distribution(struct aji_client *client);
+static iks* aji_create_pubsub_node(struct aji_client *client, const char *node_type,
+	const char *name, const char *collection_name);
+static iks* aji_build_node_config(iks *pubsub, const char *node_type, const char *collection_name);
+static void aji_create_pubsub_collection(struct aji_client *client, const char *collection_name);
+static void aji_create_pubsub_leaf(struct aji_client *client, const char *collection_name,
+   const char *leaf_name);
+static char *aji_cli_create_leafnode(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
+static void aji_create_affiliations(struct aji_client *client, const char *node);
+static iks* aji_pubsub_iq_create(struct aji_client *client, const char *type);
+static void aji_publish_device_state(struct aji_client *client, const char * device,
+	const char *device_state);
+static int aji_handle_publish_error(void *data, ikspak *pak);
+static int aji_handle_pubsub_event(void *data, ikspak *pak);
 /* No transports in this version */
 /*
 static int aji_create_transport(char *label, struct aji_client *client);
@@ -255,6 +274,9 @@
 	AST_CLI_DEFINE(aji_show_clients, "Show state of clients and components"),
 	AST_CLI_DEFINE(aji_show_buddies, "Show buddy lists of our clients"),
 	AST_CLI_DEFINE(aji_test, "Shows roster, but is generally used for mog's debugging."),
+	AST_CLI_DEFINE(aji_cli_create_collection, "Creates a test PubSub node collection."),
+	AST_CLI_DEFINE(aji_cli_list_pubsub_items, "Lists Pubsub items"),
+	AST_CLI_DEFINE(aji_cli_create_leafnode, "Creates a test Pubsub leaf node"),
 };
 
 static char *app_ajisend = "JabberSend";
@@ -467,7 +489,7 @@
 	);
 
 	if (deprecation_warning++ % 10 == 0)
-		ast_log(LOG_WARNING, "JabberStatus is deprecated.  Please use the JABBER_STATUS dialplan function in the future.\n");
+	ast_log(LOG_WARNING,"JabberStatus is deprecated Please use the JABBER_STATUS dialplan function in the future.\n");
 
 	if (!data) {
 		ast_log(LOG_ERROR, "Usage: JabberStatus(<sender>,<jid>[/<resource>],<varname>\n");
@@ -496,7 +518,8 @@
 	if (!r && buddy->resources) 
 		r = buddy->resources;
 	if (!r)
-		ast_log(LOG_NOTICE, "Resource '%s' of buddy '%s' was not found\n", jid.resource, jid.screenname);
+		ast_log(LOG_NOTICE, "Resource '%s' of buddy '%s' was not found\n", jid.resource,
+jid.screenname);
 	else
 		stat = r->status;
 	snprintf(status, sizeof(status), "%d", stat);
@@ -504,7 +527,8 @@
 	return 0;
 }
 
-static int acf_jabberstatus_read(struct ast_channel *chan, const char *name, char *data, char *buf, size_t buflen)
+static int acf_jabberstatus_read(struct ast_channel *chan, const char *name, char *data,
+char *buf, size_t buflen)
 {
 	struct aji_client *client = NULL;
 	struct aji_buddy *buddy = NULL;
@@ -545,7 +569,8 @@
 	if (!r && buddy->resources) 
 		r = buddy->resources;
 	if (!r)
-		ast_log(LOG_NOTICE, "Resource %s of buddy %s was not found.\n", jid.resource, jid.screenname);
+		ast_log(LOG_NOTICE, "Resource %s of buddy %s was not found.\n", jid.resource,
+jid.screenname);
 	else
 		stat = r->status;
 	snprintf(buf, buflen, "%d", stat);
@@ -684,7 +709,8 @@
  * \return the number of read bytes on success, 0 on timeout expiration, 
  * -1 on  error
  */
-static int aji_io_recv(struct aji_client *client, char *buffer, size_t buf_len, int timeout)
+static int aji_io_recv(struct aji_client *client, char *buffer, size_t buf_len, int
+timeout)
 {
 	int sock;
 	fd_set fds;
@@ -883,7 +909,8 @@
 	struct aji_client *client = ASTOBJ_REF((struct aji_client *) data);
 
 	if (!ast_strlen_zero(xmpp))
-		manager_event(EVENT_FLAG_USER, "JabberEvent", "Account: %s\r\nPacket: %s\r\n", client->name, xmpp);
+		manager_event(EVENT_FLAG_USER, "JabberEvent", "Account: %s\r\nPacket: %s\r\n",
+client->name, xmpp);
 
 	if (client->debug) {
 		if (is_incoming)
@@ -910,7 +937,8 @@
  *
  * \return IKS_OK on success, IKSNET_NOTSUPP on failure.
  */
-static int aji_start_sasl(struct aji_client *client, enum ikssasltype type, char *username, char *pass)
+static int aji_start_sasl(struct aji_client *client, enum ikssasltype type, char
+*username, char *pass)
 {
 	iks *x = NULL;
 	int len;
@@ -967,7 +995,8 @@
 	int features = 0;
 
 	if(!node) {
-		ast_log(LOG_ERROR, "aji_act_hook was called with out a packet\n"); /* most likely cause type is IKS_NODE_ERROR lost connection */
+		ast_log(LOG_ERROR, "aji_act_hook was called with out a packet\n"); /* most likely
+cause type is IKS_NODE_ERROR lost connection */
 		ASTOBJ_UNREF(client, aji_client_destroy);
 		return IKS_HOOK;
 	}
@@ -997,8 +1026,10 @@
 				break;
 			}
 			if (!client->usesasl) {
-				iks_filter_add_rule(client->f, aji_client_connect, client, IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, client->mid, IKS_RULE_DONE);
-				auth = jabber_make_auth(client->jid, client->password, iks_find_attrib(node, "id"));
+				iks_filter_add_rule(client->f, aji_client_connect, client, IKS_RULE_TYPE,
+IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, client->mid, IKS_RULE_DONE);
+				auth = jabber_make_auth(client->jid, client->password,
+iks_find_attrib(node, "id"));
 				if (auth) {
 					iks_insert_attrib(auth, "id", client->mid);
 					iks_insert_attrib(auth, "to", client->jid->server);
@@ -1025,7 +1056,8 @@
 						break;
 					if (client->authorized) {
 						if (features & IKS_STREAM_BIND) {
-							iks_filter_add_rule(client->f, aji_client_connect, client, IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_DONE);
+							iks_filter_add_rule(client->f, aji_client_connect, client,
+IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_DONE);
 							auth = iks_make_resource_bind(client->jid);
 							if (auth) {
 								iks_insert_attrib(auth, "id", client->mid);
@@ -1038,7 +1070,9 @@
 							}
 						}
 						if (features & IKS_STREAM_SESSION) {
-							iks_filter_add_rule (client->f, aji_client_connect, client, IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, "auth", IKS_RULE_DONE);
+							iks_filter_add_rule (client->f, aji_client_connect, client,
+IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, "auth",
+IKS_RULE_DONE);
 							auth = iks_make_session();
 							if (auth) {
 								iks_insert_attrib(auth, "id", "auth");
@@ -1056,7 +1090,8 @@
 							break;
 						}
 
-						ret = aji_start_sasl(client, features, client->jid->user, client->password);
+						ret = aji_start_sasl(client, features, client->jid->user,
+client->password);
 						if (ret != IKS_OK) {
 							ASTOBJ_UNREF(client, aji_client_destroy);
 							return IKS_HOOK;
@@ -1097,7 +1132,8 @@
 					handshake = NULL;
 				}
 				client->state = AJI_CONNECTING;
-				if(aji_recv(client, 1) == 2) /*XXX proper result for iksemel library on iks_recv of <handshake/> XXX*/
+				if(aji_recv(client, 1) == 2) /*XXX proper result for iksemel library on
+iks_recv of <handshake/> XXX*/
 					client->state = AJI_CONNECTED;
 				else
 					ast_log(LOG_WARNING, "Jabber didn't seem to handshake, failed to authenticate.\n");
@@ -1228,7 +1264,8 @@
 			iks_insert_attrib(query, "xmlns", "jabber:iq:register");
 			iks_insert_attrib(error, "code" , "406");
 			iks_insert_attrib(error, "type", "modify");
-			iks_insert_attrib(notacceptable, "xmlns", "urn:ietf:params:xml:ns:xmpp-stanzas");
+			iks_insert_attrib(notacceptable, "xmlns",
+"urn:ietf:params:xml:ns:xmpp-stanzas");
 			iks_insert_node(iq, query);
 			iks_insert_node(iq, error);
 			iks_insert_node(error, notacceptable);
@@ -1381,7 +1418,8 @@
 			ASTOBJ_UNREF(client, aji_client_destroy);
 			return IKS_FILTER_EAT;
 		}
-		if (iks_find_with_attrib(pak->query, "feature", "var", "http://www.google.com/xmpp/protocol/voice/v1")) {
+		if (iks_find_with_attrib(pak->query, "feature", "var",
+"http://www.google.com/xmpp/protocol/voice/v1")) {
 			resource->cap->jingle = 1;
 		} else
 			resource->cap->jingle = 0;
@@ -1402,7 +1440,8 @@
 			iks_insert_attrib(ident, "type", "pc");
 			iks_insert_attrib(ident, "name", "asterisk");
 			iks_insert_attrib(disco, "var", "http://jabber.org/protocol/disco#info");
-			iks_insert_attrib(google, "var", "http://www.google.com/xmpp/protocol/voice/v1");
+			iks_insert_attrib(google, "var",
+"http://www.google.com/xmpp/protocol/voice/v1");
 			iks_insert_node(iq, query);
 			iks_insert_node(query, ident);
 			iks_insert_node(query, google);
@@ -1446,12 +1485,15 @@
 			ASTOBJ_UNREF(client, aji_client_destroy);
 			return IKS_FILTER_EAT;
 		}
-		if (iks_find_with_attrib(pak->query, "feature", "var", "http://www.google.com/xmpp/protocol/voice/v1")) {
+		if (iks_find_with_attrib(pak->query, "feature", "var",
+"http://www.google.com/xmpp/protocol/voice/v1")) {
 			resource->cap->jingle = 1;
 		} else
 			resource->cap->jingle = 0;
-	} else if (pak->subtype == IKS_TYPE_GET && !(node = iks_find_attrib(pak->query, "node"))) {
-		iks *iq, *query, *identity, *disco, *reg, *commands, *gateway, *version, *vcard, *search;
+	} else if (pak->subtype == IKS_TYPE_GET && !(node = iks_find_attrib(pak->query,
+"node"))) {
+		iks *iq, *query, *identity, *disco, *reg, *commands, *gateway, *version, *vcard,
+*search;
 
 		iq = iks_new("iq");
 		query = iks_new("query");
@@ -1463,8 +1505,8 @@
 		version = iks_new("feature");
 		vcard = iks_new("feature");
 		search = iks_new("feature");
-
-		if (iq && query && identity && disco && reg && commands && gateway && version && vcard && search && client) {
+		if (iq && query && identity && disco && reg && commands && gateway && version &&
+vcard && search && client) {
 			iks_insert_attrib(iq, "from", client->user);
 			iks_insert_attrib(iq, "to", pak->from->full);
 			iks_insert_attrib(iq, "id", pak->id);
@@ -1505,8 +1547,9 @@
 		iks_delete(version);
 		iks_delete(vcard);
 		iks_delete(search);
-
-	} else if (pak->subtype == IKS_TYPE_GET && !strcasecmp(node, "http://jabber.org/protocol/commands")) {
+		
+	} else if (pak->subtype == IKS_TYPE_GET && !strcasecmp(node,
+"http://jabber.org/protocol/commands")) {
 		iks *iq, *query, *confirm;
 		iq = iks_new("iq");
 		query = iks_new("query");
@@ -1645,7 +1688,8 @@
 	}
 	ASTOBJ_WRLOCK(buddy);
 	status = (pak->show) ? pak->show : 6;
-	priority = atoi((iks_find_cdata(pak->x, "priority")) ? iks_find_cdata(pak->x, "priority") : "0");
+	priority = atoi((iks_find_cdata(pak->x, "priority")) ? iks_find_cdata(pak->x,
+"priority") : "0");
 	tmp = buddy->resources;
 	descrip = ast_strdup(iks_find_cdata(pak->x,"status"));
 
@@ -1786,7 +1830,8 @@
 				iks_insert_attrib(iq,"from", client->jid->full);
 				iks_insert_attrib(iq, "id", client->mid);
 				ast_aji_increment_mid(client->mid);
-				iks_insert_attrib(query, "xmlns", "http://jabber.org/protocol/disco#info");
+				iks_insert_attrib(query, "xmlns",
+"http://jabber.org/protocol/disco#info");
 				iks_insert_node(iq, query);
 				ast_aji_send(client, iq);
 				
@@ -1862,7 +1907,8 @@
 		iks_delete(status);
 
 		if (client->component)
-			aji_set_presence(client, pak->from->full, iks_find_attrib(pak->x, "to"), client->status, client->statusmessage);
+			aji_set_presence(client, pak->from->full, iks_find_attrib(pak->x, "to"),
+client->status, client->statusmessage);
 	case IKS_TYPE_SUBSCRIBED:
 		buddy = ASTOBJ_CONTAINER_FIND(&client->buddies, pak->from->partial);
 		if (!buddy && pak->from->partial) {
@@ -2010,6 +2056,7 @@
 		res = aji_reconnect(client);
 		sleep(4);
 	}
+	
 
 	do {
 		if (res == IKS_NET_RWERR || client->timeout == 0) {
@@ -2089,7 +2136,9 @@
 			ASTOBJ_UNLOCK(iterator);
 		});
 		iks_filter_remove_hook(client->f, aji_register_transport);
-		iks_filter_add_rule(client->f, aji_register_transport2, client, IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_NS, IKS_NS_REGISTER, IKS_RULE_DONE);
+		iks_filter_add_rule(client->f, aji_register_transport2, client, IKS_RULE_TYPE,
+IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_NS, IKS_NS_REGISTER,
+IKS_RULE_DONE);
 		iks_insert_attrib(send, "to", buddy->host);
 		iks_insert_attrib(send, "id", client->mid);
 		ast_aji_increment_mid(client->mid);
@@ -2156,7 +2205,8 @@
 #endif
 
 /*!
- * \brief goes through roster and prunes users not needed in list, or adds them accordingly.
+ * \brief goes through roster and prunes users not needed in list, or adds them
+accordingly.
  * \param client the configured XMPP client we use to connect to a XMPP server
  * \return void.
  * \note The messages here should be configurable.
@@ -2179,7 +2229,8 @@
 		ASTOBJ_RDLOCK(iterator);
 		/* For an aji_buddy, both AUTOPRUNE and AUTOREGISTER will never
 		 * be called at the same time */
-		if (ast_test_flag(&iterator->flags, AJI_AUTOPRUNE)) { /* If autoprune is set on jabber.conf */
+		if (ast_test_flag(&iterator->flags, AJI_AUTOPRUNE)) { /* If autoprune is set on
+jabber.conf */
 			res = ast_aji_send(client, iks_make_s10n(IKS_TYPE_UNSUBSCRIBE, iterator->name,
 								 "GoodBye. Your status is no longer needed by Asterisk the Open Source PBX"
 								 " so I am no longer subscribing to your presence.\n"));
@@ -2272,7 +2323,8 @@
 			if(ast_test_flag(&client->flags, AJI_AUTOPRUNE)) {
 				ast_set_flag(&buddy->flags, AJI_AUTOPRUNE);
 				ASTOBJ_MARK(buddy);
-			} else if (!iks_strcmp(iks_find_attrib(x, "subscription"), "none") || !iks_strcmp(iks_find_attrib(x, "subscription"), "from")) {
+			} else if (!iks_strcmp(iks_find_attrib(x, "subscription"), "none") ||
+!iks_strcmp(iks_find_attrib(x, "subscription"), "from")) {
 				/* subscribe to buddy's presence only 
 				   if we really need to */
 				ast_set_flag(&buddy->flags, AJI_AUTOREGISTER);
@@ -2327,7 +2379,8 @@
 
 	if(roster) {
 		iks_insert_attrib(roster, "id", "roster");
-		aji_set_presence(client, NULL, client->jid->full, client->status, client->statusmessage);
+		aji_set_presence(client, NULL, client->jid->full, client->status,
+client->statusmessage);
 		ast_aji_send(client, roster);
 	}
 
@@ -2349,12 +2402,17 @@
 
 	if (client) {
 		if (client->state == AJI_DISCONNECTED) {
-			iks_filter_add_rule(client->f, aji_filter_roster, client, IKS_RULE_TYPE, IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, "roster", IKS_RULE_DONE);
+			iks_filter_add_rule(client->f, aji_filter_roster, client, IKS_RULE_TYPE,
+IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, "roster", IKS_RULE_DONE);
 			client->state = AJI_CONNECTING;
 			client->jid = (iks_find_cdata(pak->query, "jid")) ? iks_id_new(client->stack, iks_find_cdata(pak->query, "jid")) : client->jid;
 			iks_filter_remove_hook(client->f, aji_client_connect);
-			if(!client->component) /*client*/
+			if(!client->component) { /*client*/
 				aji_get_roster(client);
+				if (client->distribute_events) {
+					aji_init_event_distribution(client);
+				}
+			}
 		}
 	} else
 		ast_log(LOG_ERROR, "Out of memory.\n");
@@ -2377,13 +2435,15 @@
 	client->stream_flags = 0;
 #endif
 	/* If it's a component, connect to user, otherwise, connect to server */
-	connected = iks_connect_via(client->p, S_OR(client->serverhost, client->jid->server), client->port, client->component ? client->user : client->jid->server);
+	connected = iks_connect_via(client->p, S_OR(client->serverhost, client->jid->server),
+client->port, client->component ? client->user : client->jid->server);
 
 	if (connected == IKS_NET_NOCONN) {
 		ast_log(LOG_ERROR, "JABBER ERROR: No Connection\n");
 		return IKS_HOOK;
 	} else 	if (connected == IKS_NET_NODNS) {
-		ast_log(LOG_ERROR, "JABBER ERROR: No DNS %s for client to  %s\n", client->name, S_OR(client->serverhost, client->jid->server));
+		ast_log(LOG_ERROR, "JABBER ERROR: No DNS %s for client to  %s\n", client->name,
+S_OR(client->serverhost, client->jid->server));
 		return IKS_HOOK;
 	}
 
@@ -2413,6 +2473,475 @@
 
 	return 1;
 }
+
+
+/*!
+ * \brief Callback function for evnts
+ * \param ast_event
+ * \param data void pointer to ast_client structure
+ * \return void
+ */
+static void aji_event_cb(const struct ast_event *ast_event, void *data)
+{
+	const char *device;
+	const char *device_state;
+	if (ast_eid_cmp(&ast_eid_default, ast_event_get_ie_raw(ast_event, AST_EVENT_IE_EID)))
+	{
+		/* If the event didn't originate from this server, don't send it back out. */
+		ast_log(LOG_DEBUG, "Returning here\n");
+		return;
+	}
+	
+	struct aji_client *client = ASTOBJ_REF((struct aji_client *) data);
+	
+	if (ast_event_get_type(ast_event) == AST_EVENT_MWI) {
+		ast_log(LOG_ERROR, "Got an MWI event, mailbox: %s\n",
+				ast_event_get_ie_str(ast_event, AST_EVENT_IE_MAILBOX));
+	} else if(ast_event_get_type(ast_event) == AST_EVENT_DEVICE_STATE_CHANGE) {
+		device = ast_event_get_ie_str(ast_event, AST_EVENT_IE_DEVICE);
+		device_state = ast_devstate_str(ast_event_get_ie_uint(ast_event, AST_EVENT_IE_STATE));
+		ast_log(LOG_ERROR, "Got a DEVICE_STATE_CHANGE event for %s\n", device);
+		aji_publish_device_state(client, device, device_state);
+		ast_log(LOG_ERROR, "DEVICE_STATE is %s\n", device_state);
+	}
+}
+
+/*!
+ * \brief Initialize collections for event distribution
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \return void
+ */
+static void aji_init_event_distribution(struct aji_client *client)
+{
+	struct ast_event_sub *mwi_sub;
+	struct ast_event_sub *device_state_sub;
+	if(ast_test_flag(&globalflags, AJI_XEP0248)) {
+		aji_create_pubsub_collection(client, "device_state");
+		aji_create_pubsub_collection(client, "message_waiting");
+	} else {
+		aji_create_pubsub_node(client, NULL, "device_state", NULL);
+		aji_create_pubsub_node(client, NULL, "message_waiting", NULL);
+
+	}
+	mwi_sub = ast_event_subscribe(AST_EVENT_MWI, aji_event_cb, "aji_mwi_subscription",
+		client, AST_EVENT_IE_END);
+	ast_log(LOG_DEBUG, "Subscribed to MWI\n");
+	iks_filter_add_rule(client->f, aji_handle_pubsub_event, client, IKS_RULE_TYPE,
+		IKS_PAK_MESSAGE, IKS_RULE_FROM, client->pubsub_node, IKS_RULE_DONE);
+	iks_filter_add_rule(client->f, aji_handle_publish_error, client, IKS_RULE_TYPE,
+		IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_ERROR, IKS_RULE_DONE);
+	
+	if (ast_enable_distributed_devstate()) {
+		return;
+	}
+	device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE_CHANGE, aji_event_cb,
+		"aji_devstate_subscription", client, AST_EVENT_IE_END);
+	ast_event_dump_cache(device_state_sub);
+	ast_log(LOG_DEBUG, "Subscribed to Devstate changes\n");
+}
+
+/*!
+ * \brief Callback for handling PubSub events
+ * \param data void pointer to aji_client structure
+ * \return IKS_FILTER_EAT
+ */
+static int aji_handle_pubsub_event(void *data, ikspak *pak)
+{
+	char *device, *device_state;
+	iks *item, *entry, *state_node;
+	struct ast_eid pubsub_eid;
+	struct ast_event *event;
+	item = iks_find(iks_find(iks_find(pak->x, "event"), "items"), "item");
+	if(item) {
+		device = iks_find_attrib(item, "id");
+		entry = iks_find(item, "entry");
+		state_node = iks_find(entry, "state");
+		if(state_node) {
+			device_state = iks_find_cdata(entry, "state");
+			ast_str_to_eid(&pubsub_eid, iks_find_attrib(state_node,"eid"));
+			if (!ast_eid_cmp(&ast_eid_default, &pubsub_eid)) {
+				ast_log(LOG_DEBUG, "Returning here, eid of incoming event matches ours!\n");
+				return IKS_FILTER_EAT;
+			}
+			if (!(event = ast_event_new(AST_EVENT_DEVICE_STATE_CHANGE,
+				AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, device, AST_EVENT_IE_STATE,
+				AST_EVENT_IE_PLTYPE_UINT, ast_devstate_val(device_state), AST_EVENT_IE_EID,
+				AST_EVENT_IE_PLTYPE_RAW, &pubsub_eid, sizeof(pubsub_eid),
+				AST_EVENT_IE_END))) {
+				return IKS_FILTER_EAT;
+			}
+
+			ast_event_queue_and_cache(event);
+		}
+	}
+	return IKS_FILTER_EAT;
+}
+
+/*!
+ * \brief Add Owner affiliations for pubsub node
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param node the name of the node to which to add affiliations
+ * \return void
+ */
+static void aji_create_affiliations(struct aji_client *client, const char *node)
+{
+	int res = 0;
+	iks *modify_affiliates = aji_pubsub_iq_create(client, "set");
+	iks *pubsub, *affiliations, *affiliate;
+	pubsub = iks_insert(modify_affiliates, "pubsub");
+	iks_insert_attrib(pubsub, "xmlns", "http://jabber.org/protocol/pubsub#owner");
+	affiliations = iks_insert(pubsub, "affiliations");
+	iks_insert_attrib(affiliations, "node", node);
+	ASTOBJ_CONTAINER_TRAVERSE(&client->buddies, 1, {
+		ASTOBJ_RDLOCK(iterator);
+		affiliate = iks_insert(affiliations, "affiliation");
+		iks_insert_attrib(affiliate, "jid", iterator->name);
+		iks_insert_attrib(affiliate, "affiliation", "owner");
+		ASTOBJ_UNLOCK(iterator);
+	});
+	res = ast_aji_send(client, modify_affiliates);
+	iks_delete(modify_affiliates);
+}
+
+/*!
+ * \brief Publish an item to a PubSub node
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param device the name of the device whose state to publish
+ * \param device_state the state to publish
+ * \return void
+ */
+static void aji_publish_device_state(struct aji_client *client, const char * device,
+	const char *device_state)
+{
+	iks *request = aji_pubsub_iq_create(client, "set");
+	iks *pubsub, *publish, *item, *entry, *state;
+	char eid_str[20];
+	ast_eid_to_str(eid_str, sizeof(eid_str), &ast_eid_default);
+	ast_log(LOG_ERROR, "eid_str is %s\n", eid_str);
+	
+	pubsub = iks_insert(request, "pubsub");
+	iks_insert_attrib(pubsub, "xmlns", "http://jabber.org/protocol/pubsub");
+	publish = iks_insert(pubsub, "publish");
+	if(ast_test_flag(&globalflags, AJI_XEP0248)) {
+		iks_insert_attrib(publish, "node", device);
+	} else {
+		iks_insert_attrib(publish, "node", "device_state");
+	}
+	item = iks_insert(publish, "item");
+	iks_insert_attrib(item, "id", device);
+	entry = iks_insert(item, "entry");
+	iks_insert_attrib(entry, "xmlns", "http://placeholder.org");
+	state = iks_insert(entry, "state");
+	iks_insert_attrib(state, "eid", eid_str);
+	iks_insert_cdata(state, device_state, strlen(device_state));
+	ast_aji_send(client, request);
+	iks_delete(request);
+}
+
+/*!
+ * \brief Create an IQ packet
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param type the type of IQ packet to create
+ * \return iks*
+ */
+static iks* aji_pubsub_iq_create(struct aji_client *client, const char *type)
+{
+	iks *request = iks_new("iq");
+	
+	iks_insert_attrib(request, "to", client->pubsub_node);
+	iks_insert_attrib(request, "from", client->jid->full);
+	iks_insert_attrib(request, "type", type);
+	ast_aji_increment_mid(client->mid);
+	iks_insert_attrib(request, "id", client->mid);
+	return request;
+}
+
+static int aji_handle_publish_error(void *data, ikspak *pak)
+{
+	char *node_name;
+	iks *publish;
+	iks *pubsub = iks_find(pak->x, "pubsub");
+	struct aji_client *client = ASTOBJ_REF((struct aji_client *) data);
+	if(pubsub) {
+		publish = iks_find(pubsub, "publish");
+		if(publish) {
+			node_name = iks_find_attrib(publish, "node");
+			if(iks_find(iks_find(iks_find(publish, "item"), "entry"), "state")) {
+				if(ast_test_flag(&globalflags, AJI_XEP0248)) {
+					aji_create_pubsub_leaf(client, "device_state", node_name);
+				} else {
+					aji_create_pubsub_node(client, NULL, node_name, NULL);
+				}
+				iks *pubsub;
+				iks *request = aji_pubsub_iq_create(client, "set");
+				pubsub = iks_insert(request, "pubsub");
+				iks_insert_attrib(pubsub, "xmlns", "http://jabber.org/protocol/pubsub");
+				iks_insert_node(pubsub, publish);
+				ast_aji_send(client, request);
+				iks_delete(request);
+				return IKS_FILTER_EAT;
+			}
+		} 
+	}
+	return IKS_FILTER_EAT;
+}
+
+/*!
+ * \brief Request item list from pubsub
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \return void
+ */
+static void aji_request_pubsub_items(struct aji_client *client)
+{
+	int res = 0;
+	iks *request = aji_pubsub_iq_create(client, "get");
+	iks *query;
+	query = iks_insert(request, "query");
+	iks_insert_attrib(query, "xmlns", "http://jabber.org/protocol/disco#items");
+	iks_filter_add_rule(client->f, aji_receive_item_list, client, IKS_RULE_TYPE,
+		IKS_PAK_IQ, IKS_RULE_SUBTYPE, IKS_TYPE_RESULT, IKS_RULE_ID, client->mid,
+		IKS_RULE_DONE);
+	res = ast_aji_send(client, request);
+	iks_delete(request);
+
+}
+
+
+/*!
+ * \brief Receive pubsub item lists
+ * \param data pointer to aji_client structure
+ * \param pak response from pubsub diso#items query
+ * \return IKS_FILTER_EAT
+ */
+static int aji_receive_item_list(void *data, ikspak* pak)
+{
+
+	struct aji_client *client = ASTOBJ_REF((struct aji_client *) data);
+	iks *item;
+	if(iks_has_children(pak->query)) {	
+		item = iks_first_tag(pak->query);
+		ast_log(LOG_WARNING, "Node name: %s\n", iks_find_attrib(item, "node"));
+		while((item = iks_next_tag(item))) {
+			ast_log(LOG_WARNING, "Node name: %s\n", iks_find_attrib(item, "node"));
+		} 
+	}
+	iks_delete(item);
+	return IKS_FILTER_EAT;
+}
+		
+	
+
+/*!
+ * \brief Method to expose PubSub node list via CLI.
+ * \return char *
+ */
+static char *aji_cli_list_pubsub_items(struct ast_cli_entry *e, int cmd, struct
+ast_cli_args *a)
+{
+		struct aji_client *client;
+		const char *name = "asterisk";
+
+		switch (cmd) {
+		case CLI_INIT:
+				e->command = "jabber list items";
+				e->usage =
+					"Usage: jabber list items [name]\n"
+					"       Lists root items on Pubsub server\n"
+					"       as configured in jabber.conf.\n";
+			return NULL;
+		case CLI_GENERATE:
+			return NULL;
+		}
+
+		if (a->argc > 4) {
+			return CLI_SHOWUSAGE;
+		} else if (a->argc == 4) {
+			name = a->argv[3];
+		}
+
+        if (!(client = ASTOBJ_CONTAINER_FIND(&clients, name))) {
+			ast_cli(a->fd, "Unable to find client '%s'!\n", name);
+			return CLI_FAILURE;
+		}
+		
+		ast_cli(a->fd, "Listing pubsub items.\n");
+		aji_request_pubsub_items(client);
+		return CLI_SUCCESS;
+}
+
+/*!
+ * \brief Create a PubSub collection node.
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param collection_name The name to use for this collection
+ * \return void.
+ */
+
+static void aji_create_pubsub_collection(struct aji_client *client, const char
+*collection_name)
+{
+	aji_create_pubsub_node(client, "collection", collection_name, NULL);
+}
+
+
+/*!
+ * \brief Create a PubSub leaf node.
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param leaf_name The name to use for this collection
+ * \return void.
+ */
+static void aji_create_pubsub_leaf(struct aji_client *client, const char *collection_name,
+const char *leaf_name)
+{
+	aji_create_pubsub_node(client, "leaf", leaf_name, collection_name);
+}
+
+/*!
+ * \brief Create a pubsub node
+ * \param client the configured XMPP client we use to connect to a XMPP server
+ * \param node_type the type of node to create
+ * \param name the name of the node to create
+ * \return iks*
+ */
+static iks* aji_create_pubsub_node(struct aji_client *client, const char *node_type, const
+		char *name, const char *collection_name)
+{
+	int res = 0;
+	iks *node = aji_pubsub_iq_create(client, "set");
+	iks *pubsub, *create, *configure;
+	pubsub = iks_insert(node, "pubsub");
+	iks_insert_attrib(pubsub, "xmlns", "http://jabber.org/protocol/pubsub");
+	create = iks_insert(pubsub, "create");
+	iks_insert_attrib(create, "node", name);
+	configure = aji_build_node_config(pubsub, node_type, collection_name);
+	res = ast_aji_send(client, node);
+	aji_create_affiliations(client, name);
+	iks_delete(node);
+	return 0;
+}
+
+
+
+static iks* aji_build_node_config(iks *pubsub, const char *node_type, const char *collection_name)
+{
+	iks *configure, *x, *field_owner, *field_node_type, *field_node_config,
+ 	*field_persist_items, *field_access_model, *field_pubsub_collection;
+	configure = iks_insert(pubsub, "configure");
+	x = iks_insert(configure, "x");
+	iks_insert_attrib(x, "xmlns", "jabber:x:data");
+	iks_insert_attrib(x, "type", "submit");
+	field_owner = iks_insert(x, "field");
+	iks_insert_attrib(field_owner, "var", "FORM_TYPE");
+	iks_insert_attrib(field_owner, "type", "hidden");
+	iks_insert_cdata(iks_insert(field_owner, "value"),
+		"http://jabber.org/protocol/pubsub#owner", 39);
+	if(node_type) {
+		field_node_type = iks_insert(x, "field");
+		iks_insert_attrib(field_node_type, "var", "pubsub#node_type");
+		iks_insert_cdata(iks_insert(field_node_type, "value"), node_type, strlen(node_type));
+	}
+	field_node_config = iks_insert(x, "field");
+	iks_insert_attrib(field_node_config, "var", "FORM_TYPE");
+	iks_insert_attrib(field_node_config, "type", "hidden");
+	iks_insert_cdata(iks_insert(field_node_config, "value"),
+		"http://jabber.org/protocol/pubsub#node_config", 45);
+	field_persist_items = iks_insert(x, "field");
+	iks_insert_attrib(field_persist_items, "var", "pubsub#persist_items");
+	iks_insert_cdata(iks_insert(field_persist_items, "value"), "1", 1);
+	field_access_model = iks_insert(x, "field");
+	iks_insert_attrib(field_access_model, "var", "pubsub#access_model");
+	iks_insert_cdata(iks_insert(field_access_model, "value"), "whitelist", 9);
+	if(node_type && !strcasecmp(node_type, "leaf")) {
+		field_pubsub_collection = iks_insert(x, "field");
+		iks_insert_attrib(field_pubsub_collection, "var", "pubsub#collection");
+		iks_insert_cdata(iks_insert(field_pubsub_collection, "value"), collection_name,
+			strlen(collection_name));
+	}
+	return configure;
+}
+
+
+
+/*!
+ * \brief Method to expose PubSub collection node creation via CLI.
+ * \return char *.
+ */
+static char *aji_cli_create_collection(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
+{
+		struct aji_client *client;
+		const char *name = "asterisk";
+		const char *collection_name = "test_collection";
+
+		switch (cmd) {
+		case CLI_INIT:
+				e->command = "jabber create collection";
+				e->usage =
+					"Usage: jabber create collection [name] [node name]\n"
+					"       Creates a PubSub collection node using the account\n"
+					"       as configured in jabber.conf.\n";
+			return NULL;
+		case CLI_GENERATE:
+			return NULL;
+		}
+
+		if (a->argc > 5) {
+			return CLI_SHOWUSAGE;
+		} else if (a->argc == 5) {
+			name = a->argv[3];
+			collection_name = a->argv[4];
+		}
+
+        if (!(client = ASTOBJ_CONTAINER_FIND(&clients, name))) {
+			ast_cli(a->fd, "Unable to find client '%s'!\n", name);
+			return CLI_FAILURE;
+		}
+		
+		ast_cli(a->fd, "Creating test PubSub node collection.\n");
+		aji_create_pubsub_collection(client, collection_name);
+		return CLI_SUCCESS;
+}
+
+/*!
+ * \brief Method to expose PubSub leaf node creation via CLI.
+ * \return char *.
+ */
+static char *aji_cli_create_leafnode(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
+{
+	struct aji_client *client;
+	const char *name = "asterisk";
+	const char *collection_name = "test_collection";
+	const char *leaf_name = "test_leaf";
+
+	switch (cmd) {
+		case CLI_INIT:
+			e->command = "jabber create leaf";
+			e->usage =
+					"Usage: jabber create leaf [name] [collection_name] [node name]\n"
+					"       Creates a PubSub leaf node using the account\n"
+					"       as configured in jabber.conf.\n";
+			return NULL;
+		case CLI_GENERATE:
+			return NULL;
+	}
+
+	if (a->argc > 6) {
+		return CLI_SHOWUSAGE;
+	} else if (a->argc == 6) {
+		name = a->argv[3];
+		collection_name = a->argv[4];
+		leaf_name = a->argv[5];
+	}
+
+	if (!(client = ASTOBJ_CONTAINER_FIND(&clients, name))) {
+		ast_cli(a->fd, "Unable to find client '%s'!\n", name);
+		return CLI_FAILURE;
+	}
+		
+	ast_cli(a->fd, "Creating test PubSub node collection.\n");
+	aji_create_pubsub_leaf(client, collection_name, leaf_name);
+	return CLI_SUCCESS;
+}
+
+
 
 /*!
  * \brief set presence of client.
@@ -2710,6 +3239,7 @@
 	client->keepalive = 1;
 	client->timeout = 50;
 	client->message_timeout = 100;
+	client->distribute_events = 0;
 	AST_LIST_HEAD_INIT(&client->messages);
 	client->component = 0;
 	ast_copy_string(client->statusmessage, "Online and Available", sizeof(client->statusmessage));
@@ -2736,9 +3266,24 @@
 		else if (!strcasecmp(var->name, "debug"))
 			client->debug = (ast_false(var->value)) ? 0 : 1;
 		else if (!strcasecmp(var->name, "type")) {
-			if (!strcasecmp(var->value, "component"))
+			if (!strcasecmp(var->value, "component")) {
 				client->component = 1;
-		} else if (!strcasecmp(var->name, "usetls")) {
+				if (client->distribute_events) {
+					ast_log(LOG_ERROR, "Client cannot be configure to be both a component and to distribute events!  Event distribution will be disabled.\n");
+					client->distribute_events = 0;
+				}
+			}
+		} else if (!strcasecmp(var->name, "distribute_events")) {
+			if(ast_true(var->value)) {
+				if (client->component) {
+					ast_log(LOG_ERROR, "Client cannot be configure to be both a component and to distribute events!  Event distribution will be disabled.\n");
+				} else {
+					client->distribute_events= 1;
+				}
+			}
+		} else if (!strcasecmp(var->name, "pubsub_node")) {
+			ast_copy_string(client->pubsub_node, var->value, sizeof(client->pubsub_node));
+		}else if (!strcasecmp(var->name, "usetls")) {
 			client->usetls = (ast_false(var->value)) ? 0 : 1;
 		} else if (!strcasecmp(var->name, "usesasl")) {
 			client->usesasl = (ast_false(var->value)) ? 0 : 1;
@@ -2832,11 +3377,14 @@
 	} else {
 		iks_filter_add_rule(client->f, aji_client_info_handler, client, IKS_RULE_NS, "http://jabber.org/protocol/disco#info", IKS_RULE_DONE);
 	}
+
 	iks_set_log_hook(client->p, aji_log_hook);
 	ASTOBJ_UNLOCK(client);
 	ASTOBJ_CONTAINER_LINK(&clients,client);
 	return 1;
 }
+
+
 
 #if 0
 /*!
@@ -2949,6 +3497,8 @@
 			ast_set2_flag(&globalflags, ast_true(var->value), AJI_AUTOPRUNE);
 		} else if (!strcasecmp(var->name, "autoregister")) {
 			ast_set2_flag(&globalflags, ast_true(var->value), AJI_AUTOREGISTER);
+		} else if (!strcasecmp(var->name, "collection_nodes")) {
+			ast_set2_flag(&globalflags, ast_true(var->value), AJI_XEP0248);
 		}
 	}
 
@@ -3060,8 +3610,13 @@
 		if(iterator->state == AJI_DISCONNECTED) {
 			if (!iterator->thread)
 				ast_pthread_create_background(&iterator->thread, NULL, aji_recv_loop, iterator);
-		} else if (iterator->state == AJI_CONNECTING)
+		} else if (iterator->state == AJI_CONNECTING) {
 			aji_get_roster(iterator);
+			if (iterator->distribute_events) {
+				aji_init_event_distribution(iterator);
+			}
+		}
+	
 		ASTOBJ_UNLOCK(iterator);
 	});
 	




More information about the asterisk-commits mailing list