class DBQueueManager extends QueueManager
{
/**
- * Saves an object into the queue item table.
+ * Saves an object reference into the queue item table.
* @return boolean true on success
* @throws ServerException on failure
*/
{
$qi = new Queue_item();
- $qi->frame = serialize($object);
+ $qi->frame = $this->encode($object);
$qi->transport = $queue;
$qi->created = common_sql_now();
$result = $qi->insert();
- if (!$result) {
+ if ($result === false) {
common_log_db_error($qi, 'INSERT', __FILE__);
throw new ServerException('DB error inserting queue item');
}
}
/**
- * Poll every minute for new events during idle periods.
+ * Poll every 10 seconds for new events during idle periods.
* We'll look in more often when there's data available.
*
* @return int seconds
*/
public function pollInterval()
{
- return 60;
+ return 10;
}
/**
* Run a polling cycle during idle processing in the input loop.
- * @return boolean true if we had a hit
+ * @return boolean true if we should poll again for more data immediately
*/
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...');
+ $qi = Queue_item::top($this->activeQueues());
+ if (empty($qi)) {
+ //$this->_log(LOG_DEBUG, 'No notices waiting; idling.');
return false;
}
- if ($qi === true) {
- // We dequeued an entry for a deleted or invalid object.
- // 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);
+ try {
+ $item = $this->decode($qi->frame);
+ } catch (Exception $e) {
+ $this->_log(LOG_INFO, "[$queue] Discarding: ".$e->getMessage());
+ $this->_done($qi);
+ return true;
+ }
- // Yay! Got one!
+ $rep = $this->logrep($item);
+ $this->_log(LOG_DEBUG, "Got $rep for transport $queue");
+
$handler = $this->getHandler($queue);
if ($handler) {
- if ($handler->handle($object)) {
- $this->_log(LOG_INFO, "[$queue] Successfully handled object");
+ if ($handler->handle($item)) {
+ $this->_log(LOG_INFO, "[$queue:$rep] Successfully handled item");
$this->_done($qi);
} else {
- $this->_log(LOG_INFO, "[$queue] Failed to handle object");
+ $this->_log(LOG_INFO, "[$queue:$rep] Failed to handle item");
$this->_fail($qi);
}
} else {
- $this->_log(LOG_INFO, "[$queue] No handler for queue $queue; discarding.");
+ $this->_log(LOG_INFO, "[$queue:$rep] 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 Queue_item
- */
- protected function _nextItem()
- {
- $start = time();
- $result = null;
-
- $qi = Queue_item::top();
- if (empty($qi)) {
- return false;
- }
-
- return $qi;
- }
-
/**
* Delete our claimed item from the queue after successful processing.
*
*/
protected function _done($qi)
{
- if (empty($qi)) {
- $this->_log(LOG_INFO, "_done passed an empty queue item");
- } else {
- if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item");
- }
- $qi->delete();
- $qi->free();
+ $queue = $qi->transport;
+
+ if (empty($qi->claimed)) {
+ $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item $qi->id from $qi->queue");
}
+ $qi->delete();
- $this->_log(LOG_INFO, "done with item");
+ $this->stats('handled', $queue);
}
/**
*/
protected function _fail($qi)
{
- if (empty($qi)) {
- $this->_log(LOG_INFO, "_fail passed an empty queue item");
+ $queue = $qi->transport;
+
+ if (empty($qi->claimed)) {
+ $this->_log(LOG_WARNING, "[$queue:item $qi->id] Ignoring failure for unclaimed queue item");
} else {
- if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "Ignoring failure for unclaimed queue item");
- } else {
- $orig = clone($qi);
- $qi->claimed = null;
- $qi->update($orig);
- $qi = null;
- }
+ $qi->releaseClaim();
}
- $this->_log(LOG_INFO, "done with queue item");
- }
-
- protected function _log($level, $msg)
- {
- common_log($level, 'DBQueueManager: '.$msg);
+ $this->stats('error', $queue);
}
}