Will re-enable anything queueing after 0.9.x merge

Revert "Any object (not just Notice's) can be queued"

This reverts commit 77ea02cac3576f395e4548e7e6cbace90ba566ea.

19 files changed, 291 insertions, 109 deletions

diff --git a/classes/Queue_item.php b/classes/Queue_item.php
index 4d90e1d23..cf805a606 100644
--- a/classes/Queue_item.php
+++ b/classes/Queue_item.php
@@ -10,8 +10,7 @@ class Queue_item extends Memcached_DataObject
     /* the code below is auto generated do not remove the above tag */
 
     public $__table = 'queue_item';                      // table name
-    public $id;                              // int(4)  primary_key not_null
-    public $frame;                           // blob   not_null
+    public $notice_id;                       // int(4)  primary_key not_null
     public $transport;                       // varchar(8)  primary_key not_null
     public $created;                         // datetime()   not_null
     public $claimed;                         // datetime()
@@ -23,6 +22,9 @@ class Queue_item extends Memcached_DataObject
     /* 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();
@@ -40,7 +42,7 @@ class Queue_item extends Memcached_DataObject
             # 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 id=' . $qi->id .
+            common_log(LOG_INFO, 'claiming queue item = ' . $qi->notice_id .
                 ' for transport ' . $qi->transport);
             $orig = clone($qi);
             $qi->claimed = common_sql_now();
@@ -55,4 +57,9 @@ class Queue_item extends Memcached_DataObject
         $qi = null;
         return null;
     }
+
+    function pkeyGet($kv)
+    {
+        return Memcached_DataObject::pkeyGet('Queue_item', $kv);
+    }
 }
diff --git a/classes/statusnet.ini b/classes/statusnet.ini
index 6203650a6..44088cf6b 100644
--- a/classes/statusnet.ini
+++ b/classes/statusnet.ini
@@ -428,14 +428,14 @@ tagged = K
 tag = K
 
 [queue_item]
-id = 129
-frame = 66
+notice_id = 129
 transport = 130
 created = 142
 claimed = 14
 
 [queue_item__keys]
-id = K
+notice_id = K
+transport = K
 
 [related_group]
 group_id = 129
diff --git a/db/statusnet.sql b/db/statusnet.sql
index cb7dad3e2..2a9ab74c7 100644
--- a/db/statusnet.sql
+++ b/db/statusnet.sql
@@ -274,12 +274,13 @@ create table remember_me (
 ) ENGINE=InnoDB CHARACTER SET utf8 COLLATE utf8_bin;
 
 create table queue_item (
-    id integer auto_increment primary key comment 'unique identifier',
-    frame blob not null comment 'serialized object',
+
+    notice_id integer not null comment 'notice queued' references notice (id),
     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;
diff --git a/lib/dbqueuemanager.php b/lib/dbqueuemanager.php
index 139f50234..889365b64 100644
--- a/lib/dbqueuemanager.php
+++ b/lib/dbqueuemanager.php
@@ -31,17 +31,19 @@
 class DBQueueManager extends QueueManager
 {
     /**
-     * Saves an object into the queue item table.
+     * Saves a notice object reference 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->frame     = serialize($object);
+        $qi->notice_id = $notice->id;
         $qi->transport = $queue;
-        $qi->created   = common_sql_now();
+        $qi->created   = $notice->created;
         $result        = $qi->insert();
 
         if (!$result) {
@@ -71,35 +73,34 @@ class DBQueueManager extends QueueManager
      */
     public function poll()
     {
-        $this->_log(LOG_DEBUG, 'Checking for queued objects...');
-        $qi = $this->_nextItem();
-        if ($qi === false) {
-            $this->_log(LOG_DEBUG, 'No queue items waiting; idling.');
+        $this->_log(LOG_DEBUG, 'Checking for notices...');
+        $item = $this->_nextItem();
+        if ($item === false) {
+            $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
             return false;
         }
-        if ($qi === true) {
-            // We dequeued an entry for a deleted or invalid object.
+        if ($item === true) {
+            // We dequeued an entry for a deleted or invalid notice.
             // Consider it a hit for poll rate purposes.
             return true;
         }
 
-        $queue = $qi->transport;
-        $object = unserialize($qi->frame);
-        $this->_log(LOG_INFO, 'Got item id=' . $qi->id . ' for transport ' . $queue);
+        list($queue, $notice) = $item;
+        $this->_log(LOG_INFO, 'Got notice '. $notice->id . ' for transport ' . $queue);
 
         // Yay! Got one!
         $handler = $this->getHandler($queue);
         if ($handler) {
-            if ($handler->handle($object)) {
-                $this->_log(LOG_INFO, "[$queue] Successfully handled object");
-                $this->_done($qi);
+            if ($handler->handle_notice($notice)) {
+                $this->_log(LOG_INFO, "[$queue:notice $notice->id] Successfully handled notice");
+                $this->_done($notice, $queue);
             } else {
-                $this->_log(LOG_INFO, "[$queue] Failed to handle object");
-                $this->_fail($qi);
+                $this->_log(LOG_INFO, "[$queue:notice $notice->id] Failed to handle notice");
+                $this->_fail($notice, $queue);
             }
         } else {
-            $this->_log(LOG_INFO, "[$queue] No handler for queue $queue; discarding.");
-            $this->_done($qi);
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] No handler for queue $queue; discarding.");
+            $this->_done($notice, $queue);
         }
         return true;
     }
@@ -107,7 +108,8 @@ class DBQueueManager extends QueueManager
     /**
      * Pop the oldest unclaimed item off the queue set and claim it.
      *
-     * @return mixed false if no items; true if bogus hit; otherwise Queue_item
+     * @return mixed false if no items; true if bogus hit; otherwise array(string, Notice)
+     *               giving the queue transport name.
      */
     protected function _nextItem()
     {
@@ -119,42 +121,70 @@ class DBQueueManager extends QueueManager
             return false;
         }
 
-        return $qi;
+        $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);
     }
 
     /**
      * Delete our claimed item from the queue after successful processing.
      *
-     * @param QueueItem $qi
+     * @param Notice $object
+     * @param string $queue
      */
-    protected function _done($qi)
+    protected function _done($object, $queue)
     {
+        // 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, "_done passed an empty queue item");
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
         } else {
             if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item");
+                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Reluctantly releasing unclaimed queue item");
             }
             $qi->delete();
             $qi->free();
         }
 
-        $this->_log(LOG_INFO, "done with item");
+        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with item");
+        $this->stats('handled', $queue);
+
+        $notice->free();
     }
 
     /**
      * Free our claimed queue item for later reprocessing in case of
      * temporary failure.
      *
-     * @param QueueItem $qi
+     * @param Notice $object
+     * @param string $queue
      */
-    protected function _fail($qi)
+    protected function _fail($object, $queue)
     {
+        // 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, "_fail passed an empty queue item");
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
         } else {
             if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, "Ignoring failure for unclaimed queue item");
+                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Ignoring failure for unclaimed queue item");
             } else {
                 $orig = clone($qi);
                 $qi->claimed = null;
@@ -163,7 +193,10 @@ class DBQueueManager extends QueueManager
             }
         }
 
-        $this->_log(LOG_INFO, "done with queue item");
+        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with queue item");
+        $this->stats('error', $queue);
+
+        $notice->free();
     }
 
     protected function _log($level, $msg)
diff --git a/lib/jabberqueuehandler.php b/lib/jabberqueuehandler.php
index 83471f2df..b1518866d 100644
--- a/lib/jabberqueuehandler.php
+++ b/lib/jabberqueuehandler.php
@@ -34,14 +34,14 @@ class JabberQueueHandler extends QueueHandler
         return 'jabber';
     }
 
-    function handle($notice)
+    function handle_notice($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());
-            return false;
+            exit(1);
         }
     }
 }
diff --git a/lib/ombqueuehandler.php b/lib/ombqueuehandler.php
index 24896c784..3ffc1313b 100644
--- a/lib/ombqueuehandler.php
+++ b/lib/ombqueuehandler.php
@@ -36,7 +36,7 @@ class OmbQueueHandler extends QueueHandler
      * @fixme doesn't currently report failure back to the queue manager
      * because omb_broadcast_notice() doesn't report it to us
      */
-    function handle($notice)
+    function handle_notice($notice)
     {
         if ($this->is_remote($notice)) {
             $this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
diff --git a/lib/pingqueuehandler.php b/lib/pingqueuehandler.php
index 4e4d74cb1..8bb218078 100644
--- a/lib/pingqueuehandler.php
+++ b/lib/pingqueuehandler.php
@@ -30,7 +30,7 @@ class PingQueueHandler extends QueueHandler {
         return 'ping';
     }
 
-    function handle($notice) {
+    function handle_notice($notice) {
         require_once INSTALLDIR . '/lib/ping.php';
         return ping_broadcast_notice($notice);
     }
diff --git a/lib/pluginqueuehandler.php b/lib/pluginqueuehandler.php
index 9653ccad4..24d504699 100644
--- a/lib/pluginqueuehandler.php
+++ b/lib/pluginqueuehandler.php
@@ -42,7 +42,7 @@ class PluginQueueHandler extends QueueHandler
         return 'plugin';
     }
 
-    function handle($notice)
+    function handle_notice($notice)
     {
         Event::handle('HandleQueuedNotice', array(&$notice));
         return true;
diff --git a/lib/publicqueuehandler.php b/lib/publicqueuehandler.php
index c9edb8d5d..9ea9ee73a 100644
--- a/lib/publicqueuehandler.php
+++ b/lib/publicqueuehandler.php
@@ -23,6 +23,7 @@ if (!defined('STATUSNET') && !defined('LACONICA')) {
 
 /**
  * Queue handler for pushing new notices to public XMPP subscribers.
+ * @fixme correct this exception handling
  */
 class PublicQueueHandler extends QueueHandler
 {
@@ -32,14 +33,15 @@ class PublicQueueHandler extends QueueHandler
         return 'public';
     }
 
-    function handle($notice)
+    function handle_notice($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());
-            return false;
+            die($e->getMessage());
         }
+        return true;
     }
 }
diff --git a/lib/queuehandler.php b/lib/queuehandler.php
index 2909cd83b..613be6e33 100644
--- a/lib/queuehandler.php
+++ b/lib/queuehandler.php
@@ -22,20 +22,51 @@ if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 /**
  * Base class for queue handlers.
  *
- * 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.
+ * 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.
  *
  * Subclasses must override at least the following methods:
  * - transport
- * - handle
+ * - handle_notice
  */
+#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.
@@ -52,17 +83,61 @@ class QueueHandler
 
     /**
      * Here's the meat of your queue handler -- you're handed a Notice
-     * or other object, which you may do as you will with.
+     * 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 mixed $object
+     * @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.
+     *
      * @return boolean true on success, false on failure
      */
-    function handle($object)
+    function run()
     {
+        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);
+    }
 }
 
diff --git a/lib/smsqueuehandler.php b/lib/smsqueuehandler.php
index 6085d2b4a..48a96409d 100644
--- a/lib/smsqueuehandler.php
+++ b/lib/smsqueuehandler.php
@@ -31,7 +31,7 @@ class SmsQueueHandler extends QueueHandler
         return 'sms';
     }
 
-    function handle($notice)
+    function handle_notice($notice)
     {
     	require_once(INSTALLDIR.'/lib/mail.php');
         return mail_broadcast_notice_sms($notice);
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index 6496b5cf1..00590fdb6 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -125,25 +125,28 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Saves an object into the queue item table.
+     * Saves a notice object reference into the queue item table.
      * @return boolean true on success
      */
     public function enqueue($object, $queue)
     {
-        $msg = serialize($object);
+        $notice = $object;
 
         $this->_connect();
 
+        // XXX: serialize and send entire notice
+
         $result = $this->con->send($this->queueName($queue),
-                                   $msg, 		// BODY of the message
-                                   array ('created' => $timestamp));
+                                   $notice->id, 		// BODY of the message
+                                   array ('created' => $notice->created));
 
         if (!$result) {
             common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
             return false;
         }
 
-        common_log(LOG_DEBUG, "complete remote queueing $log for $queue");
+        common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
+                   . $notice->id . ' for ' . $queue);
         $this->stats('enqueued', $queue);
     }
 
@@ -171,7 +174,7 @@ class StompQueueManager extends QueueManager
         $ok = true;
         $frames = $this->con->readFrames();
         foreach ($frames as $frame) {
-            $ok = $ok && $this->_handleItem($frame);
+            $ok = $ok && $this->_handleNotice($frame);
         }
         return $ok;
     }
@@ -262,10 +265,10 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Handle and acknowledge an event that's come in through a queue.
+     * Handle and acknowledge a notice event that's come in through a queue.
      *
      * If the queue handler reports failure, the message is requeued for later.
-     * Missing objects or handler classes will drop the message.
+     * Missing notices 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.
@@ -273,15 +276,24 @@ class StompQueueManager extends QueueManager
      * @param StompFrame $frame
      * @return bool
      */
-    protected function _handleItem($frame)
+    protected function _handleNotice($frame)
     {
         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
         if ($site != common_config('site', 'server')) {
             $this->stats('switch');
             StatusNet::init($site);
         }
-        $info = "object posted at {$frame->headers['created']} in queue $queue";
-        $item = unserialize($frame->body);
+
+        $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;
+        }
 
         $handler = $this->getHandler($queue);
         if (!$handler) {
@@ -291,7 +303,7 @@ class StompQueueManager extends QueueManager
             return false;
         }
 
-        $ok = $handler->handle($item);
+        $ok = $handler->handle_notice($notice);
 
         if (!$ok) {
             $this->_log(LOG_WARNING, "Failed handling $info");
@@ -299,7 +311,7 @@ class StompQueueManager extends QueueManager
             // this kind of queue management ourselves;
             // if we don't ack, it should resend...
             $this->con->ack($frame);
-            $this->enqueue($item, $queue);
+            $this->enqueue($notice, $queue);
             $this->stats('requeued', $queue);
             return false;
         }
diff --git a/lib/xmppmanager.php b/lib/xmppmanager.php
index c49986854..dfff63a30 100644
--- a/lib/xmppmanager.php
+++ b/lib/xmppmanager.php
@@ -176,30 +176,6 @@ class XmppManager extends IoManager
     }
 
     /**
-     * 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.
      */
     protected function sendPing()
diff --git a/plugins/Enjit/enjitqueuehandler.php b/plugins/Enjit/enjitqueuehandler.php
index 14085cc5e..f0e706b92 100644
--- a/plugins/Enjit/enjitqueuehandler.php
+++ b/plugins/Enjit/enjitqueuehandler.php
@@ -32,7 +32,14 @@ class EnjitQueueHandler extends QueueHandler
         return 'enjit';
     }
 
-    function handle($notice)
+    function start()
+    {
+        $this->log(LOG_INFO, "Starting EnjitQueueHandler");
+        $this->log(LOG_INFO, "Broadcasting to ".common_config('enjit', 'apiurl'));
+        return true;
+    }
+
+    function handle_notice($notice)
     {
 
         $profile = Profile::staticGet($notice->profile_id);
diff --git a/plugins/Facebook/facebookqueuehandler.php b/plugins/Facebook/facebookqueuehandler.php
index 524af7bc4..1778690e5 100644
--- a/plugins/Facebook/facebookqueuehandler.php
+++ b/plugins/Facebook/facebookqueuehandler.php
@@ -28,7 +28,7 @@ class FacebookQueueHandler extends QueueHandler
         return 'facebook';
     }
 
-    function handle($notice)
+    function handle_notice($notice)
     {
         if ($this->_isLocal($notice)) {
             return facebookBroadcastNotice($notice);
diff --git a/plugins/RSSCloud/RSSCloudPlugin.php b/plugins/RSSCloud/RSSCloudPlugin.php
index 9f444c8bb..2de162628 100644
--- a/plugins/RSSCloud/RSSCloudPlugin.php
+++ b/plugins/RSSCloud/RSSCloudPlugin.php
@@ -138,9 +138,6 @@ class RSSCloudPlugin extends Plugin
         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/' .
@@ -197,6 +194,32 @@ class RSSCloudPlugin extends Plugin
     }
 
     /**
+     * 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
      *
      * @param Notice $notice the notice in question
@@ -238,15 +261,19 @@ class RSSCloudPlugin extends Plugin
     }
 
     /**
-     * Register RSSCloud notice queue handler
+     * Add RSSCloudQueueHandler to the list of valid daemons to
+     * start
      *
-     * @param QueueManager $manager
+     * @param array $daemons the list of daemons to run
      *
      * @return boolean hook return
+     *
      */
-    function onEndInitializeQueueManager($manager)
+
+    function onGetValidDaemons($daemons)
     {
-        $manager->connect('rsscloud', 'RSSCloudQueueHandler');
+        array_push($daemons, INSTALLDIR .
+                   '/plugins/RSSCloud/RSSCloudQueueHandler.php');
         return true;
     }
 
diff --git a/plugins/RSSCloud/RSSCloudQueueHandler.php b/plugins/RSSCloud/RSSCloudQueueHandler.php
index 295c26189..693dd27c1 100644..100755
--- a/plugins/RSSCloud/RSSCloudQueueHandler.php
+++ b/plugins/RSSCloud/RSSCloudQueueHandler.php
@@ -1,3 +1,4 @@
+#!/usr/bin/env php
 <?php
 /*
  * StatusNet - the distributed open-source microblogging tool
@@ -17,20 +18,61 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
+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';
 
 class RSSCloudQueueHandler extends QueueHandler
 {
+    var $notifier = null;
+
     function transport()
     {
         return 'rsscloud';
     }
 
-    function handle($notice)
+    function start()
+    {
+        $this->log(LOG_INFO, "INITIALIZE");
+        $this->notifier = new RSSCloudNotifier();
+        return true;
+    }
+
+    function handle_notice($notice)
     {
         $profile = $notice->getProfile();
-        $notifier = new RSSCloudNotifier();
-        return $notifier->notify($profile);
+        return $this->notifier->notify($profile);
+    }
+
+    function finish()
+    {
     }
+
+}
+
+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();
diff --git a/plugins/TwitterBridge/twitterqueuehandler.php b/plugins/TwitterBridge/twitterqueuehandler.php
index b5a624e83..5089ca7b7 100644
--- a/plugins/TwitterBridge/twitterqueuehandler.php
+++ b/plugins/TwitterBridge/twitterqueuehandler.php
@@ -28,7 +28,7 @@ class TwitterQueueHandler extends QueueHandler
         return 'twitter';
     }
 
-    function handle($notice)
+    function handle_notice($notice)
     {
         return broadcast_twitter($notice);
     }
diff --git a/scripts/handlequeued.php b/scripts/handlequeued.php
index 815884969..9031437aa 100755
--- a/scripts/handlequeued.php
+++ b/scripts/handlequeued.php
@@ -50,7 +50,7 @@ if (empty($notice)) {
     exit(1);
 }
 
-if (!$handler->handle($notice)) {
+if (!$handler->handle_notice($notice)) {
     print "Failed to handle notice id $noticeId on queue '$queue'.\n";
     exit(1);
 }