X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=lib%2Fdbqueuemanager.php;h=0fc7305c0afa8daf6721136ae7e273bd22e9ffc0;hb=a7df79ac073d3d6dad13cc0fa3cd1224c9bde933;hp=3dda9fd1ac8dd14865ef4ccfee486663e73b07e4;hpb=cd29d3d646379aa9a1352035973c8e379cc7f42b;p=quix0rs-gnu-social.git

diff --git a/lib/dbqueuemanager.php b/lib/dbqueuemanager.php
index 3dda9fd1ac..0fc7305c0a 100644
--- a/lib/dbqueuemanager.php
+++ b/lib/dbqueuemanager.php
@@ -44,7 +44,7 @@ class DBQueueManager extends QueueManager
         $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');
         }
@@ -71,55 +71,71 @@ class DBQueueManager extends QueueManager
      */
     public function poll()
     {
-        $this->_log(LOG_DEBUG, 'Checking for notices...');
-        $qi = Queue_item::top($this->activeQueues());
-        if (empty($qi)) {
-            $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
+        //$this->_log(LOG_DEBUG, 'Checking for notices...');
+        $qi = Queue_item::top($this->activeQueues(), $this->getIgnoredTransports());
+        if (!$qi instanceof Queue_item) {
+            //$this->_log(LOG_DEBUG, 'No notices waiting; idling.');
             return false;
         }
 
-        $queue = $qi->transport;
-        $item = $this->decode($qi->frame);
+        try {
+            $item = $this->decode($qi->frame);
+        } catch (Exception $e) {
+            $this->_log(LOG_INFO, "[{$qi->transport}] Discarding: "._ve($e->getMessage()));
+            $this->_done($qi);
+            return true;
+        }
+
+        $rep = $this->logrep($item);
+        $this->_log(LOG_DEBUG, 'Got '._ve($rep).' for transport '._ve($qi->transport));
+        
+        try {
+            $handler = $this->getHandler($qi->transport);
+            $result = $handler->handle($item);
+        } catch (NoQueueHandlerException $e) {
+            $this->noHandlerFound($qi, $rep);
+            return true;
+        } catch (NoResultException $e) {
+            $this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('._ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
+            $result = true;
+        } catch (AlreadyFulfilledException $e) {
+            $this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('._ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
+            $result = true;
+        } catch (Exception $e) {
+            $this->_log(LOG_ERR, "[{$qi->transport}:$rep] Exception (".get_class($e).') thrown: '._ve($e->getMessage()));
+            $result = false;
+        }
 
-        if ($item) {
-            $rep = $this->logrep($item);
-            $this->_log(LOG_INFO, "Got $rep for transport $queue");
-            
-            $handler = $this->getHandler($queue);
-            if ($handler) {
-                if ($handler->handle($item)) {
-                    $this->_log(LOG_INFO, "[$queue:$rep] Successfully handled item");
-                    $this->_done($qi);
-                } else {
-                    $this->_log(LOG_INFO, "[$queue:$rep] Failed to handle item");
-                    $this->_fail($qi);
-                }
-            } else {
-                $this->_log(LOG_INFO, "[$queue:$rep] No handler for queue $queue; discarding.");
-                $this->_done($qi);
-            }
+        if ($result) {
+            $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Successfully handled item");
+            $this->_done($qi);
         } else {
-            $this->_log(LOG_INFO, "[$queue] Got empty/deleted item, discarding");
+            $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Failed to handle item");
             $this->_fail($qi);
         }
         return true;
     }
 
+    // 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);
+    }
+
     /**
      * Delete our claimed item from the queue after successful processing.
      *
      * @param QueueItem $qi
      */
-    protected function _done($qi)
+    protected function _done(Queue_item $qi)
     {
-        $queue = $qi->transport;
-
         if (empty($qi->claimed)) {
-            $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item $qi->id from $qi->queue");
+            $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item {$qi->id} from {$qi->transport}");
         }
         $qi->delete();
 
-        $this->stats('handled', $queue);
+        $this->stats('handled', $qi->transport);
     }
 
     /**
@@ -128,16 +144,16 @@ class DBQueueManager extends QueueManager
      *
      * @param QueueItem $qi
      */
-    protected function _fail($qi)
+    protected function _fail(Queue_item $qi, $releaseOnly=false)
     {
-        $queue = $qi->transport;
-
         if (empty($qi->claimed)) {
-            $this->_log(LOG_WARNING, "[$queue:item $qi->id] Ignoring failure for unclaimed queue item");
+            $this->_log(LOG_WARNING, "[{$qi->transport}:item {$qi->id}] Ignoring failure for unclaimed queue item");
         } else {
             $qi->releaseClaim();
         }
 
-        $this->stats('error', $queue);
+        if (!$releaseOnly) {
+            $this->stats('error', $qi->transport);
+        }
     }
 }