queuemanager.php 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380
  1. <?php
  2. /**
  3. * StatusNet, the distributed open-source microblogging tool
  4. *
  5. * Abstract class for i/o managers
  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 Sarven Capadisli <csarven@status.net>
  26. * @author Brion Vibber <brion@status.net>
  27. * @copyright 2009-2010 StatusNet, Inc.
  28. * @license http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  29. * @link http://status.net/
  30. */
  31. /**
  32. * Completed child classes must implement the enqueue() method.
  33. *
  34. * For background processing, classes should implement either socket-based
  35. * input (handleInput(), getSockets()) or idle-loop polling (idle()).
  36. */
  37. abstract class QueueManager extends IoManager
  38. {
  39. static $qm = null;
  40. protected $master = null;
  41. protected $handlers = array();
  42. protected $groups = array();
  43. protected $activeGroups = array();
  44. protected $ignoredTransports = array();
  45. /**
  46. * Factory function to pull the appropriate QueueManager object
  47. * for this site's configuration. It can then be used to queue
  48. * events for later processing or to spawn a processing loop.
  49. *
  50. * Plugins can add to the built-in types by hooking StartNewQueueManager.
  51. *
  52. * @return QueueManager
  53. */
  54. public static function get()
  55. {
  56. if (empty(self::$qm)) {
  57. if (Event::handle('StartNewQueueManager', array(&self::$qm))) {
  58. $enabled = common_config('queue', 'enabled');
  59. $type = common_config('queue', 'subsystem');
  60. if (!$enabled) {
  61. // does everything immediately
  62. self::$qm = new UnQueueManager();
  63. } else {
  64. switch ($type) {
  65. case 'db':
  66. self::$qm = new DBQueueManager();
  67. break;
  68. case 'stomp':
  69. self::$qm = new StompQueueManager();
  70. break;
  71. default:
  72. throw new ServerException("No queue manager class for type '$type'");
  73. }
  74. }
  75. }
  76. }
  77. return self::$qm;
  78. }
  79. /**
  80. * @fixme wouldn't necessarily work with other class types.
  81. * Better to change the interface...?
  82. */
  83. public static function multiSite()
  84. {
  85. if (common_config('queue', 'subsystem') == 'stomp') {
  86. return IoManager::INSTANCE_PER_PROCESS;
  87. } else {
  88. return IoManager::SINGLE_ONLY;
  89. }
  90. }
  91. function __construct()
  92. {
  93. $this->initialize();
  94. }
  95. /**
  96. * Optional; ping any running queue handler daemons with a notification
  97. * such as announcing a new site to handle or requesting clean shutdown.
  98. * This avoids having to restart all the daemons manually to update configs
  99. * and such.
  100. *
  101. * Called from scripts/queuectl.php controller utility.
  102. *
  103. * @param string $event event key
  104. * @param string $param optional parameter to append to key
  105. * @return boolean success
  106. */
  107. public function sendControlSignal($event, $param='')
  108. {
  109. throw new Exception(get_class($this) . " does not support control signals.");
  110. }
  111. /**
  112. * Store an object (usually/always a Notice) into the given queue
  113. * for later processing. No guarantee is made on when it will be
  114. * processed; it could be immediately or at some unspecified point
  115. * in the future.
  116. *
  117. * Must be implemented by any queue manager.
  118. *
  119. * @param Notice $object
  120. * @param string $queue
  121. */
  122. abstract function enqueue($object, $queue);
  123. /**
  124. * Build a representation for an object for logging
  125. * @param mixed
  126. * @return string
  127. */
  128. function logrep($object) {
  129. if (is_object($object)) {
  130. $class = get_class($object);
  131. if (isset($object->id)) {
  132. return "$class $object->id";
  133. }
  134. return $class;
  135. } elseif (is_string($object)) {
  136. $len = strlen($object);
  137. $fragment = mb_substr($object, 0, 32);
  138. if (mb_strlen($object) > 32) {
  139. $fragment .= '...';
  140. }
  141. return "string '$fragment' ($len bytes)";
  142. } elseif (is_array($object)) {
  143. return 'array with ' . count($object) .
  144. ' elements (keys:[' . implode(',', array_keys($object)) . '])';
  145. }
  146. return strval($object);
  147. }
  148. /**
  149. * Encode an object for queued storage.
  150. *
  151. * @param mixed $item
  152. * @return string
  153. */
  154. protected function encode($item)
  155. {
  156. return serialize($item);
  157. }
  158. /**
  159. * Decode an object from queued storage.
  160. * Accepts notice reference entries and serialized items.
  161. *
  162. * @param string
  163. * @return mixed
  164. */
  165. protected function decode($frame)
  166. {
  167. $object = unserialize($frame);
  168. // If it is a string, we really store a JSON object in there
  169. // except if it begins with '<', because then it is XML.
  170. if (is_string($object) &&
  171. substr($object, 0, 1) != '<' &&
  172. !is_numeric($object))
  173. {
  174. $json = json_decode($object);
  175. if ($json === null) {
  176. throw new Exception('Bad frame in queue item');
  177. }
  178. // The JSON object has a type parameter which contains the class
  179. if (empty($json->type)) {
  180. throw new Exception('Type not specified for queue item');
  181. }
  182. if (!is_a($json->type, 'Managed_DataObject', true)) {
  183. throw new Exception('Managed_DataObject class does not exist for queue item');
  184. }
  185. // And each of these types should have a unique id (or uri)
  186. if (isset($json->id) && !empty($json->id)) {
  187. $object = call_user_func(array($json->type, 'getKV'), 'id', $json->id);
  188. } elseif (isset($json->uri) && !empty($json->uri)) {
  189. $object = call_user_func(array($json->type, 'getKV'), 'uri', $json->uri);
  190. }
  191. // But if no object was found, there's nothing we can handle
  192. if (!$object instanceof Managed_DataObject) {
  193. throw new Exception('Queue item frame referenced a non-existant object');
  194. }
  195. }
  196. // If the frame was not a string, it's either an array or an object.
  197. return $object;
  198. }
  199. /**
  200. * Instantiate the appropriate QueueHandler class for the given queue.
  201. *
  202. * @param string $queue
  203. * @return mixed QueueHandler or null
  204. */
  205. function getHandler($queue)
  206. {
  207. if (isset($this->handlers[$queue])) {
  208. $class = $this->handlers[$queue];
  209. if(is_object($class)) {
  210. return $class;
  211. } else if (class_exists($class)) {
  212. return new $class();
  213. } else {
  214. $this->_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
  215. }
  216. }
  217. throw new NoQueueHandlerException($queue);
  218. }
  219. /**
  220. * Get a list of registered queue transport names to be used
  221. * for listening in this daemon.
  222. *
  223. * @return array of strings
  224. */
  225. function activeQueues()
  226. {
  227. $queues = array();
  228. foreach ($this->activeGroups as $group) {
  229. if (isset($this->groups[$group])) {
  230. $queues = array_merge($queues, $this->groups[$group]);
  231. }
  232. }
  233. return array_keys($queues);
  234. }
  235. function getIgnoredTransports()
  236. {
  237. return array_keys($this->ignoredTransports);
  238. }
  239. function ignoreTransport($transport)
  240. {
  241. // key is used for uniqueness, value doesn't mean anything
  242. $this->ignoredTransports[$transport] = true;
  243. }
  244. /**
  245. * Initialize the list of queue handlers for the current site.
  246. *
  247. * @event StartInitializeQueueManager
  248. * @event EndInitializeQueueManager
  249. */
  250. function initialize()
  251. {
  252. $this->handlers = array();
  253. $this->groups = array();
  254. $this->groupsByTransport = array();
  255. if (Event::handle('StartInitializeQueueManager', array($this))) {
  256. $this->connect('distrib', 'DistribQueueHandler');
  257. $this->connect('ping', 'PingQueueHandler');
  258. if (common_config('sms', 'enabled')) {
  259. $this->connect('sms', 'SmsQueueHandler');
  260. }
  261. // Background user management tasks...
  262. $this->connect('deluser', 'DelUserQueueHandler');
  263. $this->connect('feedimp', 'FeedImporter');
  264. $this->connect('actimp', 'ActivityImporter');
  265. $this->connect('acctmove', 'AccountMover');
  266. $this->connect('actmove', 'ActivityMover');
  267. // For compat with old plugins not registering their own handlers.
  268. $this->connect('plugin', 'PluginQueueHandler');
  269. }
  270. Event::handle('EndInitializeQueueManager', array($this));
  271. }
  272. /**
  273. * Register a queue transport name and handler class for your plugin.
  274. * Only registered transports will be reliably picked up!
  275. *
  276. * @param string $transport
  277. * @param string $class class name or object instance
  278. * @param string $group
  279. */
  280. public function connect($transport, $class, $group='main')
  281. {
  282. $this->handlers[$transport] = $class;
  283. $this->groups[$group][$transport] = $class;
  284. $this->groupsByTransport[$transport] = $group;
  285. }
  286. /**
  287. * Set the active group which will be used for listening.
  288. * @param string $group
  289. */
  290. function setActiveGroup($group)
  291. {
  292. $this->activeGroups = array($group);
  293. }
  294. /**
  295. * Set the active group(s) which will be used for listening.
  296. * @param array $groups
  297. */
  298. function setActiveGroups($groups)
  299. {
  300. $this->activeGroups = $groups;
  301. }
  302. /**
  303. * @return string queue group for this queue
  304. */
  305. function queueGroup($queue)
  306. {
  307. if (isset($this->groupsByTransport[$queue])) {
  308. return $this->groupsByTransport[$queue];
  309. } else {
  310. throw new Exception("Requested group for unregistered transport $queue");
  311. }
  312. }
  313. /**
  314. * Send a statistic ping to the queue monitoring system,
  315. * optionally with a per-queue id.
  316. *
  317. * @param string $key
  318. * @param string $queue
  319. */
  320. function stats($key, $queue=false)
  321. {
  322. $owners = array();
  323. if ($queue) {
  324. $owners[] = "queue:$queue";
  325. $owners[] = "site:" . common_config('site', 'server');
  326. }
  327. if (isset($this->master)) {
  328. $this->master->stats($key, $owners);
  329. } else {
  330. $monitor = new QueueMonitor();
  331. $monitor->stats($key, $owners);
  332. }
  333. }
  334. protected function _log($level, $msg)
  335. {
  336. $class = get_class($this);
  337. if ($this->activeGroups) {
  338. $groups = ' (' . implode(',', $this->activeGroups) . ')';
  339. } else {
  340. $groups = '';
  341. }
  342. common_log($level, "$class$groups: $msg");
  343. }
  344. }