modules/msgqueue/msgqueue.module

<?php

// $Id: $

/**
 * @file msgqueue module
 * The "msgqueue" module integrates Drupal with messaging system
 * like Apache ActiveMQ. This module accesses Drupal events by
 * implementing hooks for users, nodes, and comments, and is
 * licensed under GPLv2. The hooks supply events when any insert,
 * update, or delete action takes place for any user, node or comment.
 * An XML message about any such Drupal event is sent to the
 * messaging system. The module formats XML messages using minixml,
 * and puts them into a queue via the Stomp protocol of ActiveMQ.
 *
 * @author Larry Hamel, Codeguild.com
*/

require_once 'stomp_sender.inc'; # communication protocol bridging to ActiveMQ message broker; requires php_sockets.dll on windows
require_once 'minixml_impexp/minixml.inc.php'; # xml library

# message queue properties
# todo: make configurations?
define(MSGQUEUE_LOGIN, "hiram");
define(MSGQUEUE_PASS, "test");
define(MSGQUEUE_HOST, "localhost");
define(MSGQUEUE_QUEUENAME, "TEST.FOO");

/**
 * Implementation of hook_comment
 */
function msgqueue_comment($comment, $op) {
  if (in_array($op, array('insert', 'update', 'delete'))) {

  	// send to message queue
    if (is_array($comment)) {
      $comment = (object)$comment;
    }

    msgqueue_send('comment', $op, $comment);
  }
}

/**
 * Implementation of hook_nodeapi
 */
function msgqueue_nodeapi(&$node, $op, $teaser = NULL, $page = NULL) {
  if (in_array($op, array('insert', 'update', 'delete'))) {
    msgqueue_send('node', $op, $node);
  }
}

/**
 * Implementation of hook_user
 */
function msgqueue_user($op, $edit, $account, $category = NULL) {
  //we don't support updates for other categories than 'account'
  if ($op == 'update' && $category != 'account') {
    return;
  }
  else if (in_array($op, array('insert', 'update', 'delete'))) {
    msgqueue_send('user', $op, $account);
  }
}

/**
 * send a message to queue via Stomp protocol
 * @param itemType one of ['comment', 'node', 'user'] to indicate the source of the event
 * @param op one of ['update', 'insert', 'delete'] to indicate the operation
 * @param the object or array that represents the item (comment, node, or user) being acted upon
 */
function msgqueue_send($itemtype, $op, $obj) {
	$conn = new StompConn(MSGQUEUE_HOST);
	# user/pass not required for sample server; included here for reference (however, login() call always necessary)
	$conn->login(MSGQUEUE_LOGIN, MSGQUEUE_PASS);

	$msg = msgqueue_format($itemtype, $op, $obj);
	$conn->send("/queue/" . MSGQUEUE_QUEUENAME, $msg);
}

/**
 * convert the incoming event to an xml message.
 * @param itemType one of ['comment', 'node', 'user'] to indicate the source of the event
 * @param op one of ['update', 'insert', 'delete'] to indicate the operation
 * @param the object or array that represents the item (comment, node, or user) being acted upon
 */
function msgqueue_format($itemtype, $op, $obj) {
	$xmlDoc = new MiniXMLDoc();
	$xmlRoot =& $xmlDoc->getRoot();

	$msgElm =& $xmlRoot->createChild('msg');
	$itemElm =& $msgElm->createChild($itemtype);
	$itemElm->attribute('op', $op);

	switch ($itemtype) {
		case "comment":
			$comment = $obj;
			# for consistent handling, make sure we have object format
		    if (is_array($obj)) {
		      $comment = (object)$obj;
		    }

			$itemElm->attribute('cid', $comment->cid);
			$itemElm->attribute('nid', $comment->nid);
			$itemElm->attribute('pid', $comment->pid);
			$itemElm->attribute('uid', $comment->uid);

			if ( $op != "delete") {
				msgqueue_addtextchild($itemElm, "subject", $comment->subject);
				msgqueue_addtextchild($itemElm, "body", $comment->comment);
			}
			break;

		case "user":
			$user = $obj;
			$itemElm->attribute('uid', $user->uid);

			if ( $op != "delete") {
				$itemElm->attribute('name', $user->name);
				// todo: can mail have chars not allowed in xml?
				$itemElm->attribute('mail', $user->mail);
				$itemElm->attribute('mode', $user->mode);
				$itemElm->attribute('created', timestamp2iso($user->created));
				$itemElm->attribute('access', $user->access);
				$itemElm->attribute('login', $user->login);
				$itemElm->attribute('status', $user->status);
				$itemElm->attribute('timezone', $user->timezone);
				$itemElm->attribute('language', $user->language);
			}
			break;

		case "node":
			$node = $obj;
			$itemElm->attribute('nid', $node->nid);

			if ( $op != "delete") {
				$itemElm->attribute('vid', $node->vid);
				$itemElm->attribute('type', $node->type);
				$itemElm->attribute('language', $node->language);
				$itemElm->attribute('uid', $node->uid);
				$itemElm->attribute('status', $node->status);
				$itemElm->attribute('created', timestamp2iso($node->created));
				$itemElm->attribute('changed', timestamp2iso($node->changed));
				$itemElm->attribute('comment', $node->comment);
				$itemElm->attribute('promote', $node->promote);
				$itemElm->attribute('sticky', $node->sticky);
				$itemElm->attribute('tnid', $node->tnid);
				$itemElm->attribute('translate', $node->translate);

				msgqueue_addtextchild($itemElm, "title", $node->title);
				msgqueue_addtextchild($itemElm, "body", $node->body);
			}
			break;
	}

	return $xmlDoc->toString();
}

/**
 * convert timestamp to iso 8601
 */
 function timestamp2iso($timestamp) {
 	return date('Y-m-d\TH:i:sO', $timestamp);
 }

/**
 * use minixml to add a child element with the provided text
 * @param parent the parent element
 * @param elmname the new child element's name
 * @param val the text to put in the new element
 *
 */
function msgqueue_addtextchild($parent, $elmname, $val) {
	$myElm = $parent->createChild($elmname);
	$myElm->text(msgqueue_xmlcdata($val));
	return $myElm;
}

/**
 * add CDATA brackets to text
 */
function msgqueue_xmlcdata($txt) {
	return '<![CDATA[' . $txt . ']]>';
}
1	<?php
2
3	// $Id: $
4
5	/**
6	* @file msgqueue module
7	* The "msgqueue" module integrates Drupal with messaging system
8	* like Apache ActiveMQ. This module accesses Drupal events by
9	* implementing hooks for users, nodes, and comments, and is
10	* licensed under GPLv2. The hooks supply events when any insert,
11	* update, or delete action takes place for any user, node or comment.
12	* An XML message about any such Drupal event is sent to the
13	* messaging system. The module formats XML messages using minixml,
14	* and puts them into a queue via the Stomp protocol of ActiveMQ.
15	*
16	* @author Larry Hamel, Codeguild.com
17	*/
18
19	require_once 'stomp_sender.inc'; # communication protocol bridging to ActiveMQ message broker; requires php_sockets.dll on windows
20	require_once 'minixml_impexp/minixml.inc.php'; # xml library
21
22	# message queue properties
23	# todo: make configurations?
24	define(MSGQUEUE_LOGIN, "hiram");
25	define(MSGQUEUE_PASS, "test");
26	define(MSGQUEUE_HOST, "localhost");
27	define(MSGQUEUE_QUEUENAME, "TEST.FOO");
28
29	/**
30	* Implementation of hook_comment
31	*/
32	function msgqueue_comment($comment, $op) {
33	if (in_array($op, array('insert', 'update', 'delete'))) {
34
35	// send to message queue
36	if (is_array($comment)) {
37	$comment = (object)$comment;
38	}
39
40	msgqueue_send('comment', $op, $comment);
41	}
42	}
43
44	/**
45	* Implementation of hook_nodeapi
46	*/
47	function msgqueue_nodeapi(&$node, $op, $teaser = NULL, $page = NULL) {
48	if (in_array($op, array('insert', 'update', 'delete'))) {
49	msgqueue_send('node', $op, $node);
50	}
51	}
52
53	/**
54	* Implementation of hook_user
55	*/
56	function msgqueue_user($op, $edit, $account, $category = NULL) {
57	//we don't support updates for other categories than 'account'
58	if ($op == 'update' && $category != 'account') {
59	return;
60	}
61	else if (in_array($op, array('insert', 'update', 'delete'))) {
62	msgqueue_send('user', $op, $account);
63	}
64	}
65
66	/**
67	* send a message to queue via Stomp protocol
68	* @param itemType one of ['comment', 'node', 'user'] to indicate the source of the event
69	* @param op one of ['update', 'insert', 'delete'] to indicate the operation
70	* @param the object or array that represents the item (comment, node, or user) being acted upon
71	*/
72	function msgqueue_send($itemtype, $op, $obj) {
73	$conn = new StompConn(MSGQUEUE_HOST);
74	# user/pass not required for sample server; included here for reference (however, login() call always necessary)
75	$conn->login(MSGQUEUE_LOGIN, MSGQUEUE_PASS);
76
77	$msg = msgqueue_format($itemtype, $op, $obj);
78	$conn->send("/queue/" . MSGQUEUE_QUEUENAME, $msg);
79	}
80
81	/**
82	* convert the incoming event to an xml message.
83	* @param itemType one of ['comment', 'node', 'user'] to indicate the source of the event
84	* @param op one of ['update', 'insert', 'delete'] to indicate the operation
85	* @param the object or array that represents the item (comment, node, or user) being acted upon
86	*/
87	function msgqueue_format($itemtype, $op, $obj) {
88	$xmlDoc = new MiniXMLDoc();
89	$xmlRoot =& $xmlDoc->getRoot();
90
91	$msgElm =& $xmlRoot->createChild('msg');
92	$itemElm =& $msgElm->createChild($itemtype);
93	$itemElm->attribute('op', $op);
94
95	switch ($itemtype) {
96	case "comment":
97	$comment = $obj;
98	# for consistent handling, make sure we have object format
99	if (is_array($obj)) {
100	$comment = (object)$obj;
101	}
102
103	$itemElm->attribute('cid', $comment->cid);
104	$itemElm->attribute('nid', $comment->nid);
105	$itemElm->attribute('pid', $comment->pid);
106	$itemElm->attribute('uid', $comment->uid);
107
108	if ( $op != "delete") {
109	msgqueue_addtextchild($itemElm, "subject", $comment->subject);
110	msgqueue_addtextchild($itemElm, "body", $comment->comment);
111	}
112	break;
113
114	case "user":
115	$user = $obj;
116	$itemElm->attribute('uid', $user->uid);
117
118	if ( $op != "delete") {
119	$itemElm->attribute('name', $user->name);
120	// todo: can mail have chars not allowed in xml?
121	$itemElm->attribute('mail', $user->mail);
122	$itemElm->attribute('mode', $user->mode);
123	$itemElm->attribute('created', timestamp2iso($user->created));
124	$itemElm->attribute('access', $user->access);
125	$itemElm->attribute('login', $user->login);
126	$itemElm->attribute('status', $user->status);
127	$itemElm->attribute('timezone', $user->timezone);
128	$itemElm->attribute('language', $user->language);
129	}
130	break;
131
132	case "node":
133	$node = $obj;
134	$itemElm->attribute('nid', $node->nid);
135
136	if ( $op != "delete") {
137	$itemElm->attribute('vid', $node->vid);
138	$itemElm->attribute('type', $node->type);
139	$itemElm->attribute('language', $node->language);
140	$itemElm->attribute('uid', $node->uid);
141	$itemElm->attribute('status', $node->status);
142	$itemElm->attribute('created', timestamp2iso($node->created));
143	$itemElm->attribute('changed', timestamp2iso($node->changed));
144	$itemElm->attribute('comment', $node->comment);
145	$itemElm->attribute('promote', $node->promote);
146	$itemElm->attribute('sticky', $node->sticky);
147	$itemElm->attribute('tnid', $node->tnid);
148	$itemElm->attribute('translate', $node->translate);
149
150	msgqueue_addtextchild($itemElm, "title", $node->title);
151	msgqueue_addtextchild($itemElm, "body", $node->body);
152	}
153	break;
154	}
155
156	return $xmlDoc->toString();
157	}
158
159	/**
160	* convert timestamp to iso 8601
161	*/
162	function timestamp2iso($timestamp) {
163	return date('Y-m-d\TH:i:sO', $timestamp);
164	}
165
166	/**
167	* use minixml to add a child element with the provided text
168	* @param parent the parent element
169	* @param elmname the new child element's name
170	* @param val the text to put in the new element
171	*
172	*/
173	function msgqueue_addtextchild($parent, $elmname, $val) {
174	$myElm = $parent->createChild($elmname);
175	$myElm->text(msgqueue_xmlcdata($val));
176	return $myElm;
177	}
178
179	/**
180	* add CDATA brackets to text
181	*/
182	function msgqueue_xmlcdata($txt) {
183	return '<![CDATA[' . $txt . ']]>';
184	}