Misses this file to merge. I like the comments.

[quix0rs-gnu-social.git] / lib / dbqueuemanager.php

diff --git a/lib/dbqueuemanager.php b/lib/dbqueuemanager.php
index 139f502340d5dc84edc2d2aebafe2561306cd0e1..45c4b694d2e4848237bcae4563461fbb79ab55be 100644 (file)
--- a/lib/dbqueuemanager.php
+++ b/lib/dbqueuemanager.php
@@ -31,7 +31,7 @@
 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
      */
@@ -39,12 +39,12 @@ class DBQueueManager extends QueueManager
     {
         $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');
         }
@@ -55,71 +55,60 @@ class DBQueueManager extends QueueManager
     }
 
     /**
-     * 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 (!$qi instanceof Queue_item) {
+            //$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.
+
+        try {
+            $item = $this->decode($qi->frame);
+        } catch (Exception $e) {
+            $this->_log(LOG_INFO, "[{$qi->transport}] Discarding: ".$e->getMessage());
+            $this->_done($qi);
             return true;
         }
 
-        $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);
+        $rep = $this->logrep($item);
+        $this->_log(LOG_DEBUG, "Got {$rep} for transport {$qi->transport}");
+        
+        $handler = $this->getHandler($qi->transport);
         if ($handler) {
-            if ($handler->handle($object)) {
-                $this->_log(LOG_INFO, "[$queue] Successfully handled object");
+            if ($handler->handle($item)) {
+                $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Successfully handled item");
                 $this->_done($qi);
             } else {
-                $this->_log(LOG_INFO, "[$queue] Failed to handle object");
+                $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Failed to handle item");
                 $this->_fail($qi);
             }
         } else {
-            $this->_log(LOG_INFO, "[$queue] No handler for queue $queue; discarding.");
-            $this->_done($qi);
+            $this->noHandlerFound($qi, $rep);
         }
         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;
+    // What to do if no handler was found. For example, the OpportunisticQM
+    // should avoid deleting items just because it can't reach XMPP queues etc.
+    protected function noHandlerFound(Queue_item $qi, $rep=null) {
+        $this->_log(LOG_INFO, "[{$qi->transport}:{$rep}] No handler for queue {$qi->transport}; discarding.");
+        $this->_done($qi);
     }
 
     /**
@@ -127,19 +116,14 @@ class DBQueueManager extends QueueManager
      *
      * @param QueueItem $qi
      */
-    protected function _done($qi)
+    protected function _done(Queue_item $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();
+        if (empty($qi->claimed)) {
+            $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item {$qi->id} from {$qi->transport}");
         }
+        $qi->delete();
 
-        $this->_log(LOG_INFO, "done with item");
+        $this->stats('handled', $qi->transport);
     }
 
     /**
@@ -148,26 +132,16 @@ class DBQueueManager extends QueueManager
      *
      * @param QueueItem $qi
      */
-    protected function _fail($qi)
+    protected function _fail(Queue_item $qi, $releaseOnly=false)
     {
-        if (empty($qi)) {
-            $this->_log(LOG_INFO, "_fail passed an empty queue item");
+        if (empty($qi->claimed)) {
+            $this->_log(LOG_WARNING, "[{$qi->transport}: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);
+        if (!$releaseOnly) {
+            $this->stats('error', $qi->transport);
+        }
     }
 }