I am using modRegister as a message queue for various things (comment notifications, mailchimp sync, email change validation, etc..)
it works good and simple, and I must say it’s quite nice to have a mq built in modx.
I handle the register/unregister and the event handling itself in the models, and there’s a simple execution script (time-based cron’ed) that simply queries the queue.
About the queue read - everything (almost) works well, but something is a bit strange with the timing logic.
a. there’s an unknown delay (10-60s??) between the time messages are in the queue (i see in the mysql table) until the register read actually can pick them up.
I then refresh 2-3 times (or just wait a min..) - then register reads them.. any clue why? I am sure about this issue, as I see it all around, doesn’t matter in which model.
note: when i un-register a message and check - the read has no time gap. strange.
b. timing: for example with $options[’time_limit’] = 30; $options[’msg_limit’] = 10;
1. if the queue is empty it’ll just hang for 30sec (as expected), but it does NOT pick up any messages during that time, as it should
2. if the queue is not empty - it’ll process it and quit (without waiting 30s) even if the queue contains only 2 messages (ie. not reaching the msg_limit)
I’ve played around with this and digged in the source code, the safest option i found until now was to do a $options[’poll_limit’] = 1 (as opposed to 0 - unlimited) to force only one pass and cron this script every 2 min...
I haven’t seen any sample use anywhere which actually uses the time_limit and poll_interval, in the core - it’s always a single pass.
c. On some scripts I use the ’remove_read’ = false, as I want to control this based on the returned result (for example, in case the mail server is down, i’d retry it later)
As I use the above single-pass-fast-cron: I want to be sure this script does not execute in parallel (to avoid duplicates). can this be enforced (ie. similar to resource lock)?
here’s a sample small script:
require_once MODX_CORE_PATH.'components/activityStream/model/activityStream/activityStream.class.php';
$modx->activityStream = new ActivityStream($modx);
$options = array();
$options['msg_limit'] = 20 ;
//$options['time_limit'] = 30;
//$options['poll_interval'] = 1;
$options['poll_limit'] = 1;
$options['remove_read'] = false;
$modx->getService('registry', 'registry.modRegistry');
$modx->registry->addRegister($modx->activityStream->mqRegister,'registry.modDbRegister');
$modx->registry->notification->connect();
$modx->registry->notification->subscribe($modx->activityStream->mqTopic);
$msgs = $modx->registry->notification->read($options);
if (empty($msgs)) exit();
$output = array();
foreach ($msgs as $msg) {
if (!empty($msg)) {
$ret = $modx->activityStream->eventNotification($msg); // Handle the event and return status
if ($ret) { // OK? REMOVE_READ MESSAGES FROM MQ
$output[] = date("H:i:s") . " : SUCCESS $ret ($msg)";
$topicObj = $modx->getObject('registry.db.modDbRegisterTopic', array('name' => $modx->activityStream->mqTopic)); // get topicObj (to convert from name --> id)
$modx->removeObject('modDbRegisterMessage', array('topic' => $topicObj->get('id'), 'id' => $modx->activityStream->mqIdPrefix . $msg)); // remove message from mq
}
else { // ERROR
$output[] = date("H:i:s") ." : ERROR $ret ($msg)";
}
}
}
var_dump($output);
MODX Staff