forked from GNUsocial/gnu-social
Any object (not just Notice's) can be queued
This commit is contained in:
parent
6a93cb2985
commit
77ea02cac3
@ -10,7 +10,8 @@ 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 $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()
|
||||
@ -22,9 +23,6 @@ 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();
|
||||
@ -42,7 +40,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 = ' . $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();
|
||||
@ -57,9 +55,4 @@ class Queue_item extends Memcached_DataObject
|
||||
$qi = null;
|
||||
return null;
|
||||
}
|
||||
|
||||
function pkeyGet($kv)
|
||||
{
|
||||
return Memcached_DataObject::pkeyGet('Queue_item', $kv);
|
||||
}
|
||||
}
|
||||
|
@ -428,14 +428,14 @@ tagged = K
|
||||
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
|
||||
|
@ -274,13 +274,12 @@ create table remember_me (
|
||||
) 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;
|
||||
|
@ -31,19 +31,17 @@
|
||||
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) {
|
||||
@ -73,34 +71,35 @@ class DBQueueManager extends QueueManager
|
||||
*/
|
||||
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;
|
||||
}
|
||||
@ -108,8 +107,7 @@ 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 array(string, Notice)
|
||||
* giving the queue transport name.
|
||||
* @return mixed false if no items; true if bogus hit; otherwise Queue_item
|
||||
*/
|
||||
protected function _nextItem()
|
||||
{
|
||||
@ -121,70 +119,42 @@ class DBQueueManager extends QueueManager
|
||||
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;
|
||||
@ -193,10 +163,7 @@ class DBQueueManager extends QueueManager
|
||||
}
|
||||
}
|
||||
|
||||
$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)
|
||||
|
@ -34,14 +34,14 @@ class JabberQueueHandler extends QueueHandler
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -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($notice)
|
||||
function handle($notice)
|
||||
{
|
||||
if ($this->is_remote($notice)) {
|
||||
$this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
|
||||
|
@ -30,7 +30,7 @@ class PingQueueHandler extends QueueHandler {
|
||||
return 'ping';
|
||||
}
|
||||
|
||||
function handle_notice($notice) {
|
||||
function handle($notice) {
|
||||
require_once INSTALLDIR . '/lib/ping.php';
|
||||
return ping_broadcast_notice($notice);
|
||||
}
|
||||
|
@ -42,7 +42,7 @@ class PluginQueueHandler extends QueueHandler
|
||||
return 'plugin';
|
||||
}
|
||||
|
||||
function handle_notice($notice)
|
||||
function handle($notice)
|
||||
{
|
||||
Event::handle('HandleQueuedNotice', array(&$notice));
|
||||
return true;
|
||||
|
@ -23,7 +23,6 @@ if (!defined('STATUSNET') && !defined('LACONICA')) {
|
||||
|
||||
/**
|
||||
* Queue handler for pushing new notices to public XMPP subscribers.
|
||||
* @fixme correct this exception handling
|
||||
*/
|
||||
class PublicQueueHandler extends QueueHandler
|
||||
{
|
||||
@ -33,15 +32,14 @@ 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 true;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -22,51 +22,20 @@ if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
|
||||
/**
|
||||
* 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.
|
||||
@ -83,61 +52,17 @@ class QueueHandler
|
||||
|
||||
/**
|
||||
* 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
|
||||
* @param mixed $object
|
||||
* @return boolean true on success, false on failure
|
||||
*/
|
||||
function handle_notice($notice)
|
||||
function handle($object)
|
||||
{
|
||||
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 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);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -31,7 +31,7 @@ class SmsQueueHandler extends QueueHandler
|
||||
return 'sms';
|
||||
}
|
||||
|
||||
function handle_notice($notice)
|
||||
function handle($notice)
|
||||
{
|
||||
require_once(INSTALLDIR.'/lib/mail.php');
|
||||
return mail_broadcast_notice_sms($notice);
|
||||
|
@ -125,28 +125,25 @@ class StompQueueManager 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
|
||||
*/
|
||||
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);
|
||||
}
|
||||
|
||||
@ -174,7 +171,7 @@ class StompQueueManager extends QueueManager
|
||||
$ok = true;
|
||||
$frames = $this->con->readFrames();
|
||||
foreach ($frames as $frame) {
|
||||
$ok = $ok && $this->_handleNotice($frame);
|
||||
$ok = $ok && $this->_handleItem($frame);
|
||||
}
|
||||
return $ok;
|
||||
}
|
||||
@ -265,10 +262,10 @@ class StompQueueManager extends QueueManager
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
@ -276,24 +273,15 @@ class StompQueueManager extends QueueManager
|
||||
* @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) {
|
||||
@ -303,7 +291,7 @@ class StompQueueManager extends QueueManager
|
||||
return false;
|
||||
}
|
||||
|
||||
$ok = $handler->handle_notice($notice);
|
||||
$ok = $handler->handle($item);
|
||||
|
||||
if (!$ok) {
|
||||
$this->_log(LOG_WARNING, "Failed handling $info");
|
||||
@ -311,7 +299,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($notice, $queue);
|
||||
$this->enqueue($item, $queue);
|
||||
$this->stats('requeued', $queue);
|
||||
return false;
|
||||
}
|
||||
|
@ -175,6 +175,30 @@ 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.
|
||||
*/
|
||||
|
@ -32,14 +32,7 @@ class EnjitQueueHandler extends QueueHandler
|
||||
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);
|
||||
|
@ -28,7 +28,7 @@ class FacebookQueueHandler extends QueueHandler
|
||||
return 'facebook';
|
||||
}
|
||||
|
||||
function handle_notice($notice)
|
||||
function handle($notice)
|
||||
{
|
||||
if ($this->_isLocal($notice)) {
|
||||
return facebookBroadcastNotice($notice);
|
||||
|
@ -138,6 +138,9 @@ 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/' .
|
||||
@ -193,32 +196,6 @@ class RSSCloudPlugin extends Plugin
|
||||
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
|
||||
*
|
||||
@ -261,19 +238,15 @@ class RSSCloudPlugin extends Plugin
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
|
52
plugins/RSSCloud/RSSCloudQueueHandler.php
Executable file → Normal file
52
plugins/RSSCloud/RSSCloudQueueHandler.php
Executable file → Normal file
@ -1,4 +1,3 @@
|
||||
#!/usr/bin/env php
|
||||
<?php
|
||||
/*
|
||||
* StatusNet - the distributed open-source microblogging tool
|
||||
@ -18,61 +17,20 @@
|
||||
* 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);
|
||||
$notifier = new RSSCloudNotifier();
|
||||
return $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();
|
||||
|
@ -28,7 +28,7 @@ class TwitterQueueHandler extends QueueHandler
|
||||
return 'twitter';
|
||||
}
|
||||
|
||||
function handle_notice($notice)
|
||||
function handle($notice)
|
||||
{
|
||||
return broadcast_twitter($notice);
|
||||
}
|
||||
|
@ -50,7 +50,7 @@ if (empty($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);
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user