dbqueuemanager.php 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157
  1. <?php
  2. /**
  3. * StatusNet, the distributed open-source microblogging tool
  4. *
  5. * Simple-minded queue manager for storing items in the database
  6. *
  7. * PHP version 5
  8. *
  9. * LICENCE: This program is free software: you can redistribute it and/or modify
  10. * it under the terms of the GNU Affero General Public License as published by
  11. * the Free Software Foundation, either version 3 of the License, or
  12. * (at your option) any later version.
  13. *
  14. * This program is distributed in the hope that it will be useful,
  15. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. * GNU Affero General Public License for more details.
  18. *
  19. * You should have received a copy of the GNU Affero General Public License
  20. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  21. *
  22. * @category QueueManager
  23. * @package StatusNet
  24. * @author Evan Prodromou <evan@status.net>
  25. * @author Brion Vibber <brion@status.net>
  26. * @copyright 2009-2010 StatusNet, Inc.
  27. * @license http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  28. * @link http://status.net/
  29. */
  30. class DBQueueManager extends QueueManager
  31. {
  32. /**
  33. * Saves an object reference into the queue item table.
  34. * @return boolean true on success
  35. * @throws ServerException on failure
  36. */
  37. public function enqueue($object, $queue)
  38. {
  39. $qi = new Queue_item();
  40. $qi->frame = $this->encode($object);
  41. $qi->transport = $queue;
  42. $qi->created = common_sql_now();
  43. $result = $qi->insert();
  44. if ($result === false) {
  45. common_log_db_error($qi, 'INSERT', __FILE__);
  46. throw new ServerException('DB error inserting queue item');
  47. }
  48. $this->stats('enqueued', $queue);
  49. return true;
  50. }
  51. /**
  52. * Poll every 10 seconds for new events during idle periods.
  53. * We'll look in more often when there's data available.
  54. *
  55. * @return int seconds
  56. */
  57. public function pollInterval()
  58. {
  59. return 10;
  60. }
  61. /**
  62. * Run a polling cycle during idle processing in the input loop.
  63. * @return boolean true if we should poll again for more data immediately
  64. */
  65. public function poll()
  66. {
  67. //$this->_log(LOG_DEBUG, 'Checking for notices...');
  68. $qi = Queue_item::top($this->activeQueues(), $this->getIgnoredTransports());
  69. if (!$qi instanceof Queue_item) {
  70. //$this->_log(LOG_DEBUG, 'No notices waiting; idling.');
  71. return false;
  72. }
  73. try {
  74. $item = $this->decode($qi->frame);
  75. } catch (Exception $e) {
  76. $this->_log(LOG_INFO, "[{$qi->transport}] Discarding: ".$e->getMessage());
  77. $this->_done($qi);
  78. return true;
  79. }
  80. $rep = $this->logrep($item);
  81. $this->_log(LOG_DEBUG, "Got {$rep} for transport {$qi->transport}");
  82. try {
  83. $handler = $this->getHandler($qi->transport);
  84. $result = $handler->handle($item);
  85. } catch (NoQueueHandlerException $e) {
  86. $this->noHandlerFound($qi, $rep);
  87. return true;
  88. } catch (AlreadyFulfilledException $e) {
  89. $this->_log(LOG_ERR, "[{$qi->transport}:$rep] AlreadyFulfilledException thrown: {$e->getMessage()}");
  90. $result = true;
  91. } catch (Exception $e) {
  92. $this->_log(LOG_ERR, "[{$qi->transport}:$rep] Exception thrown: {$e->getMessage()}");
  93. $result = false;
  94. }
  95. if ($result) {
  96. $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Successfully handled item");
  97. $this->_done($qi);
  98. } else {
  99. $this->_log(LOG_INFO, "[{$qi->transport}:$rep] Failed to handle item");
  100. $this->_fail($qi);
  101. }
  102. return true;
  103. }
  104. // What to do if no handler was found. For example, the OpportunisticQM
  105. // should avoid deleting items just because it can't reach XMPP queues etc.
  106. protected function noHandlerFound(Queue_item $qi, $rep=null) {
  107. $this->_log(LOG_INFO, "[{$qi->transport}:{$rep}] No handler for queue {$qi->transport}; discarding.");
  108. $this->_done($qi);
  109. }
  110. /**
  111. * Delete our claimed item from the queue after successful processing.
  112. *
  113. * @param QueueItem $qi
  114. */
  115. protected function _done(Queue_item $qi)
  116. {
  117. if (empty($qi->claimed)) {
  118. $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item {$qi->id} from {$qi->transport}");
  119. }
  120. $qi->delete();
  121. $this->stats('handled', $qi->transport);
  122. }
  123. /**
  124. * Free our claimed queue item for later reprocessing in case of
  125. * temporary failure.
  126. *
  127. * @param QueueItem $qi
  128. */
  129. protected function _fail(Queue_item $qi, $releaseOnly=false)
  130. {
  131. if (empty($qi->claimed)) {
  132. $this->_log(LOG_WARNING, "[{$qi->transport}:item {$qi->id}] Ignoring failure for unclaimed queue item");
  133. } else {
  134. $qi->releaseClaim();
  135. }
  136. if (!$releaseOnly) {
  137. $this->stats('error', $qi->transport);
  138. }
  139. }
  140. }