123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160 |
- <?php
- class DBQueueManager extends QueueManager
- {
-
- public function enqueue($object, $queue)
- {
- $qi = new Queue_item();
- $qi->frame = $this->encode($object);
- $qi->transport = $queue;
- $qi->created = common_sql_now();
- $result = $qi->insert();
- if ($result === false) {
- common_log_db_error($qi, 'INSERT', __FILE__);
- throw new ServerException('DB error inserting queue item');
- }
- $this->stats('enqueued', $queue);
- return true;
- }
-
- public function pollInterval()
- {
- return 10;
- }
-
- public function poll()
- {
-
- $qi = Queue_item::top($this->activeQueues(), $this->getIgnoredTransports());
- if (!$qi instanceof Queue_item) {
-
- return false;
- }
- 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 ($result) {
- $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Successfully handled item");
- $this->_done($qi);
- } else {
- $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Failed to handle item");
- $this->_fail($qi);
- }
- return true;
- }
-
-
- 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);
- }
-
- protected function _done(Queue_item $qi)
- {
- if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item {$qi->id} from {$qi->transport}");
- }
- $qi->delete();
- $this->stats('handled', $qi->transport);
- }
-
- protected function _fail(Queue_item $qi, $releaseOnly=false)
- {
- if (empty($qi->claimed)) {
- $this->_log(LOG_WARNING, "[{$qi->transport}:item {$qi->id}] Ignoring failure for unclaimed queue item");
- } else {
- $qi->releaseClaim();
- }
- if (!$releaseOnly) {
- $this->stats('error', $qi->transport);
- }
- }
- }
|