/* the code below is auto generated do not remove the above tag */
public $__table = 'queue_item'; // table name
- public $notice_id; // int(4) primary_key not_null
+ public $id; // int(4) primary_key not_null
+ public $frame; // blob not_null
public $transport; // varchar(8) primary_key not_null
public $created; // datetime() not_null
public $claimed; // datetime()
/* the code above is auto generated do not remove the tag below */
###END_AUTOCODE
- function sequenceKey()
- { return array(false, false); }
-
static function top($transport=null) {
$qi = new Queue_item();
# XXX: potential race condition
# can we force it to only update if claimed is still null
# (or old)?
- common_log(LOG_INFO, 'claiming queue item = ' . $qi->notice_id .
+ common_log(LOG_INFO, 'claiming queue item id=' . $qi->id .
' for transport ' . $qi->transport);
$orig = clone($qi);
$qi->claimed = common_sql_now();
$qi = null;
return null;
}
-
- function pkeyGet($kv)
- {
- return Memcached_DataObject::pkeyGet('Queue_item', $kv);
- }
}
tag = K
[queue_item]
-notice_id = 129
+id = 129
+frame = 66
transport = 130
created = 142
claimed = 14
[queue_item__keys]
-notice_id = K
-transport = K
+id = K
[related_group]
group_id = 129
) ENGINE=InnoDB CHARACTER SET utf8 COLLATE utf8_bin;
create table queue_item (
-
- notice_id integer not null comment 'notice queued' references notice (id),
+ id integer auto_increment primary key comment 'unique identifier',
+ frame blob not null comment 'serialized object',
transport varchar(8) not null comment 'queue for what? "email", "jabber", "sms", "irc", ...',
created datetime not null comment 'date this record was created',
claimed datetime comment 'date this item was claimed',
- constraint primary key (notice_id, transport),
index queue_item_created_idx (created)
) ENGINE=InnoDB CHARACTER SET utf8 COLLATE utf8_bin;
class DBQueueManager extends QueueManager
{
/**
- * Saves a notice object reference into the queue item table.
+ * Saves an object into the queue item table.
* @return boolean true on success
* @throws ServerException on failure
*/
public function enqueue($object, $queue)
{
- $notice = $object;
-
$qi = new Queue_item();
- $qi->notice_id = $notice->id;
+ $qi->frame = serialize($object);
$qi->transport = $queue;
- $qi->created = $notice->created;
+ $qi->created = common_sql_now();
$result = $qi->insert();
if (!$result) {
*/
public function poll()
{
- $this->_log(LOG_DEBUG, 'Checking for notices...');
- $item = $this->_nextItem();
- if ($item === false) {
- $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
+ $this->_log(LOG_DEBUG, 'Checking for queued objects...');
+ $qi = $this->_nextItem();
+ if ($qi === false) {
+ $this->_log(LOG_DEBUG, 'No queue items waiting; idling.');
return false;
}
- if ($item === true) {
- // We dequeued an entry for a deleted or invalid notice.
+ if ($qi === true) {
+ // We dequeued an entry for a deleted or invalid object.
// Consider it a hit for poll rate purposes.
return true;
}
- list($queue, $notice) = $item;
- $this->_log(LOG_INFO, 'Got notice '. $notice->id . ' for transport ' . $queue);
+ $queue = $qi->transport;
+ $object = unserialize($qi->frame);
+ $this->_log(LOG_INFO, 'Got item id=' . $qi->id . ' for transport ' . $queue);
// Yay! Got one!
$handler = $this->getHandler($queue);
if ($handler) {
- if ($handler->handle_notice($notice)) {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] Successfully handled notice");
- $this->_done($notice, $queue);
+ if ($handler->handle($object)) {
+ $this->_log(LOG_INFO, "[$queue] Successfully handled object");
+ $this->_done($qi);
} else {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] Failed to handle notice");
- $this->_fail($notice, $queue);
+ $this->_log(LOG_INFO, "[$queue] Failed to handle object");
+ $this->_fail($qi);
}
} else {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] No handler for queue $queue; discarding.");
- $this->_done($notice, $queue);
+ $this->_log(LOG_INFO, "[$queue] No handler for queue $queue; discarding.");
+ $this->_done($qi);
}
return true;
}
/**
* Pop the oldest unclaimed item off the queue set and claim it.
*
- * @return mixed false if no items; true if bogus hit; otherwise array(string, Notice)
- * giving the queue transport name.
+ * @return mixed false if no items; true if bogus hit; otherwise Queue_item
*/
protected function _nextItem()
{
return false;
}
- $queue = $qi->transport;
- $notice = Notice::staticGet('id', $qi->notice_id);
- if (empty($notice)) {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] dequeued non-existent notice");
- $qi->delete();
- return true;
- }
-
- $result = $notice;
- return array($queue, $notice);
+ return $qi;
}
/**
* Delete our claimed item from the queue after successful processing.
*
- * @param Notice $object
- * @param string $queue
+ * @param QueueItem $qi
*/
- protected function _done($object, $queue)
+ protected function _done($qi)
{
- // XXX: right now, we only handle notices
-
- $notice = $object;
-
- $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
- 'transport' => $queue));
-
if (empty($qi)) {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
+ $this->_log(LOG_INFO, "_done passed an empty queue item");
} else {
if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Reluctantly releasing unclaimed queue item");
+ $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item");
}
$qi->delete();
$qi->free();
}
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with item");
- $this->stats('handled', $queue);
-
- $notice->free();
+ $this->_log(LOG_INFO, "done with item");
}
/**
* Free our claimed queue item for later reprocessing in case of
* temporary failure.
*
- * @param Notice $object
- * @param string $queue
+ * @param QueueItem $qi
*/
- protected function _fail($object, $queue)
+ protected function _fail($qi)
{
- // XXX: right now, we only handle notices
-
- $notice = $object;
-
- $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
- 'transport' => $queue));
-
if (empty($qi)) {
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
+ $this->_log(LOG_INFO, "_fail passed an empty queue item");
} else {
if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Ignoring failure for unclaimed queue item");
+ $this->_log(LOG_WARNING, "Ignoring failure for unclaimed queue item");
} else {
$orig = clone($qi);
$qi->claimed = null;
}
}
- $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with queue item");
- $this->stats('error', $queue);
-
- $notice->free();
+ $this->_log(LOG_INFO, "done with queue item");
}
protected function _log($level, $msg)
return 'jabber';
}
- function handle_notice($notice)
+ function handle($notice)
{
require_once(INSTALLDIR.'/lib/jabber.php');
try {
return jabber_broadcast_notice($notice);
} catch (XMPPHP_Exception $e) {
$this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
- exit(1);
+ return false;
}
}
}
* @fixme doesn't currently report failure back to the queue manager
* because omb_broadcast_notice() doesn't report it to us
*/
- function handle_notice($notice)
+ function handle($notice)
{
if ($this->is_remote($notice)) {
$this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
return 'ping';
}
- function handle_notice($notice) {
+ function handle($notice) {
require_once INSTALLDIR . '/lib/ping.php';
return ping_broadcast_notice($notice);
}
return 'plugin';
}
- function handle_notice($notice)
+ function handle($notice)
{
Event::handle('HandleQueuedNotice', array(&$notice));
return true;
/**
* Queue handler for pushing new notices to public XMPP subscribers.
- * @fixme correct this exception handling
*/
class PublicQueueHandler extends QueueHandler
{
return 'public';
}
- function handle_notice($notice)
+ function handle($notice)
{
require_once(INSTALLDIR.'/lib/jabber.php');
try {
return jabber_public_notice($notice);
} catch (XMPPHP_Exception $e) {
$this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
- die($e->getMessage());
+ return false;
}
- return true;
}
}
/**
* Base class for queue handlers.
*
- * As extensions of the Daemon class, each queue handler has the ability
- * to launch itself in the background, at which point it'll pass control
- * to the configured QueueManager class to poll for updates.
+ * As of 0.9, queue handlers are short-lived for items as they are
+ * dequeued by a QueueManager running in an IoMaster in a daemon
+ * such as queuedaemon.php.
+ *
+ * Extensions requiring long-running maintenance or polling should
+ * register an IoManager.
*
* Subclasses must override at least the following methods:
* - transport
- * - handle_notice
+ * - handle
*/
-#class QueueHandler extends Daemon
class QueueHandler
{
-# function __construct($id=null, $daemonize=true)
-# {
-# parent::__construct($daemonize);
-#
-# if ($id) {
-# $this->set_id($id);
-# }
-# }
-
- /**
- * How many seconds a polling-based queue manager should wait between
- * checks for new items to handle.
- *
- * Defaults to 60 seconds; override to speed up or slow down.
- *
- * @fixme not really compatible with global queue manager
- * @return int timeout in seconds
- */
-# function timeout()
-# {
-# return 60;
-# }
-
-# function class_name()
-# {
-# return ucfirst($this->transport()) . 'Handler';
-# }
-
-# function name()
-# {
-# return strtolower($this->class_name().'.'.$this->get_id());
-# }
-
/**
* Return transport keyword which identifies items this queue handler
* services; must be defined for all subclasses.
/**
* Here's the meat of your queue handler -- you're handed a Notice
- * object, which you may do as you will with.
+ * or other object, which you may do as you will with.
*
* If this function indicates failure, a warning will be logged
* and the item is placed back in the queue to be re-run.
*
- * @param Notice $notice
- * @return boolean true on success, false on failure
- */
- function handle_notice($notice)
- {
- return true;
- }
-
- /**
- * Setup and start of run loop for this queue handler as a daemon.
- * Most of the heavy lifting is passed on to the QueueManager's service()
- * method, which passes control back to our handle_notice() method for
- * each notice that comes in on the queue.
- *
- * Most of the time this won't need to be overridden in a subclass.
- *
+ * @param mixed $object
* @return boolean true on success, false on failure
*/
- function run()
+ function handle($object)
{
- if (!$this->start()) {
- $this->log(LOG_WARNING, 'failed to start');
- return false;
- }
-
- $this->log(LOG_INFO, 'checking for queued notices');
-
- $queue = $this->transport();
- $timeout = $this->timeout();
-
- $qm = QueueManager::get();
-
- $qm->service($queue, $this);
-
- $this->log(LOG_INFO, 'finished servicing the queue');
-
- if (!$this->finish()) {
- $this->log(LOG_WARNING, 'failed to clean up');
- return false;
- }
-
- $this->log(LOG_INFO, 'terminating normally');
-
return true;
}
-
-
- function log($level, $msg)
- {
- common_log($level, $this->class_name() . ' ('. $this->get_id() .'): '.$msg);
- }
}
return 'sms';
}
- function handle_notice($notice)
+ function handle($notice)
{
require_once(INSTALLDIR.'/lib/mail.php');
return mail_broadcast_notice_sms($notice);
}
/**
- * Saves a notice object reference into the queue item table.
+ * Saves an object into the queue item table.
* @return boolean true on success
*/
public function enqueue($object, $queue)
{
- $notice = $object;
+ $msg = serialize($object);
$this->_connect();
- // XXX: serialize and send entire notice
-
$result = $this->con->send($this->queueName($queue),
- $notice->id, // BODY of the message
- array ('created' => $notice->created));
+ $msg, // BODY of the message
+ array ('created' => $timestamp));
if (!$result) {
common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
return false;
}
- common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
- . $notice->id . ' for ' . $queue);
+ common_log(LOG_DEBUG, "complete remote queueing $log for $queue");
$this->stats('enqueued', $queue);
}
$ok = true;
$frames = $this->con->readFrames();
foreach ($frames as $frame) {
- $ok = $ok && $this->_handleNotice($frame);
+ $ok = $ok && $this->_handleItem($frame);
}
return $ok;
}
}
/**
- * Handle and acknowledge a notice event that's come in through a queue.
+ * Handle and acknowledge an event that's come in through a queue.
*
* If the queue handler reports failure, the message is requeued for later.
- * Missing notices or handler classes will drop the message.
+ * Missing objects or handler classes will drop the message.
*
* Side effects: in multi-site mode, may reset site configuration to
* match the site that queued the event.
* @param StompFrame $frame
* @return bool
*/
- protected function _handleNotice($frame)
+ protected function _handleItem($frame)
{
list($site, $queue) = $this->parseDestination($frame->headers['destination']);
if ($site != common_config('site', 'server')) {
$this->stats('switch');
StatusNet::init($site);
}
-
- $id = intval($frame->body);
- $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
-
- $notice = Notice::staticGet('id', $id);
- if (empty($notice)) {
- $this->_log(LOG_WARNING, "Skipping missing $info");
- $this->con->ack($frame);
- $this->stats('badnotice', $queue);
- return false;
- }
+ $info = "object posted at {$frame->headers['created']} in queue $queue";
+ $item = unserialize($frame->body);
$handler = $this->getHandler($queue);
if (!$handler) {
return false;
}
- $ok = $handler->handle_notice($notice);
+ $ok = $handler->handle($item);
if (!$ok) {
$this->_log(LOG_WARNING, "Failed handling $info");
// this kind of queue management ourselves;
// if we don't ack, it should resend...
$this->con->ack($frame);
- $this->enqueue($notice, $queue);
+ $this->enqueue($item, $queue);
$this->stats('requeued', $queue);
return false;
}
}
}
+ /**
+ * For queue handlers to pass us a message to push out,
+ * if we're active.
+ *
+ * @fixme should this be blocking etc?
+ *
+ * @param string $msg XML stanza to send
+ * @return boolean success
+ */
+ public function send($msg)
+ {
+ if ($this->conn && !$this->conn->isDisconnected()) {
+ $bytes = $this->conn->send($msg);
+ if ($bytes > 0) {
+ return true;
+ } else {
+ return false;
+ }
+ } else {
+ // Can't send right now...
+ return false;
+ }
+ }
+
/**
* Send a keepalive ping to the XMPP server.
*/
return 'enjit';
}
- function start()
- {
- $this->log(LOG_INFO, "Starting EnjitQueueHandler");
- $this->log(LOG_INFO, "Broadcasting to ".common_config('enjit', 'apiurl'));
- return true;
- }
-
- function handle_notice($notice)
+ function handle($notice)
{
$profile = Profile::staticGet($notice->profile_id);
return 'facebook';
}
- function handle_notice($notice)
+ function handle($notice)
{
if ($this->_isLocal($notice)) {
return facebookBroadcastNotice($notice);
case 'RSSCloudNotifier':
include_once INSTALLDIR . '/plugins/RSSCloud/RSSCloudNotifier.php';
return false;
+ case 'RSSCloudQueueHandler':
+ include_once INSTALLDIR . '/plugins/RSSCloud/RSSCloudQueueHandler.php';
+ return false;
case 'RSSCloudRequestNotifyAction':
case 'LoggingAggregatorAction':
include_once INSTALLDIR . '/plugins/RSSCloud/' .
return true;
}
- /**
- * broadcast the message when not using queuehandler
- *
- * @param Notice &$notice the notice
- * @param array $queue destination queue
- *
- * @return boolean hook return
- */
-
- function onUnqueueHandleNotice(&$notice, $queue)
- {
- if (($queue == 'rsscloud') && ($this->_isLocal($notice))) {
-
- common_debug('broadcasting rssCloud bound notice ' . $notice->id);
-
- $profile = $notice->getProfile();
-
- $notifier = new RSSCloudNotifier();
- $notifier->notify($profile);
-
- return false;
- }
-
- return true;
- }
-
/**
* Determine whether the notice was locally created
*
}
/**
- * Add RSSCloudQueueHandler to the list of valid daemons to
- * start
+ * Register RSSCloud notice queue handler
*
- * @param array $daemons the list of daemons to run
+ * @param QueueManager $manager
*
* @return boolean hook return
- *
*/
-
- function onGetValidDaemons($daemons)
+ function onEndInitializeQueueManager($manager)
{
- array_push($daemons, INSTALLDIR .
- '/plugins/RSSCloud/RSSCloudQueueHandler.php');
+ $manager->connect('rsscloud', 'RSSCloudQueueHandler');
return true;
}
-#!/usr/bin/env php
<?php
/*
* StatusNet - the distributed open-source microblogging tool
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/../..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_ENJIT_HELP
-Daemon script for pushing new notices to RSSCloud subscribers.
-
- -i --id Identity (default none)
-
-END_OF_ENJIT_HELP;
-
-require_once INSTALLDIR . '/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-require_once INSTALLDIR . '/plugins/RSSCloud/RSSCloudNotifier.php';
-require_once INSTALLDIR . '/plugins/RSSCloud/RSSCloudSubscription.php';
+if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
class RSSCloudQueueHandler extends QueueHandler
{
- var $notifier = null;
-
function transport()
{
return 'rsscloud';
}
- function start()
- {
- $this->log(LOG_INFO, "INITIALIZE");
- $this->notifier = new RSSCloudNotifier();
- return true;
- }
-
- function handle_notice($notice)
+ function handle($notice)
{
$profile = $notice->getProfile();
- return $this->notifier->notify($profile);
- }
-
- function finish()
- {
+ $notifier = new RSSCloudNotifier();
+ return $notifier->notify($profile);
}
-
-}
-
-if (have_option('i')) {
- $id = get_option_value('i');
-} else if (have_option('--id')) {
- $id = get_option_value('--id');
-} else if (count($args) > 0) {
- $id = $args[0];
-} else {
- $id = null;
}
-$handler = new RSSCloudQueueHandler($id);
-
-$handler->runOnce();
return 'twitter';
}
- function handle_notice($notice)
+ function handle($notice)
{
return broadcast_twitter($notice);
}
exit(1);
}
-if (!$handler->handle_notice($notice)) {
+if (!$handler->handle($notice)) {
print "Failed to handle notice id $noticeId on queue '$queue'.\n";
exit(1);
}
projects / quix0rs-gnu-social.git / commitdiff