diff --git a/README.txt b/README.txt index a43ada1..07f639d 100644 --- a/README.txt +++ b/README.txt @@ -1,37 +1,42 @@ DESCRIPTION ============== -Advanced Queue is an extended queuing module fully backward compatible +Advanced Queue is an extended queuing module fully backward compatible with and a drop-in replacement of DrupalQueue. Supports: -* A Drush-based execution engine for queued job, supporting executing +* A Drush-based execution engine for queued job, supporting executing multiple queues at the same time and timeouts +* A "poor-man's" processing via cron. * Human readable, translatable names for queued items * Tags for queued items -* Status of queued items (new, being processed, succeeded, failed), and +* Status of queued items (new, being processed, succeeded, failed), and result payload * Views integration In order to use Drush based execution engine your module must provide hook_advanced_queue_info() in which a processing callback (worker) must be -declared. This callback will receive the item being processed and process +declared. This callback will receive the item being processed and process it. Enable the "Advacned-queue example" module to see it in action. +Note that by default the "poor-man's" processing via cron is enabled, as it +allows site builders to start with the non-scalable and later on, more advanced +users can disable that option via admin/config/system/cron. + DRUSH WORKERS ============= It is possible to execute the Drush comamnd that will loop and "listen" to the queue, and process items once they are added. It is considered a best practice to execute the Drush command with a timeout of 15 minutes, and -use supervisord to restart it once it dies. This will help making sure +use supervisord to restart it once it dies. This will help making sure there are no serious memory leaks. For example: drush advancedqueue --all --timeout=900 SPONSOROS & MAINTAINERS ======================= -* Sponsored by Publicis Modem +* Sponsored by Publicis Modem * Developed and sponsored by Commerce Guys http://commerceguys.com/ * Developed by Gizra http://www.gizra.com/ diff --git a/advancedqueue.info b/advancedqueue.info index 7697e17..ace3a37 100644 --- a/advancedqueue.info +++ b/advancedqueue.info @@ -9,6 +9,3 @@ files[] = advancedqueue.queue.inc files[] = views/advancedqueue_handler_field_title.inc files[] = views/advancedqueue_handler_field_status.inc files[] = views/advancedqueue_handler_filter_status.inc - -; Test -files[] = tests/advancedqueue.test diff --git a/advancedqueue.module b/advancedqueue.module index 16c9450..396b31b 100644 --- a/advancedqueue.module +++ b/advancedqueue.module @@ -26,16 +26,6 @@ define('ADVANCEDQUEUE_STATUS_SUCCESS', 1); define('ADVANCEDQUEUE_STATUS_FAILURE', 2); /** - * Implements hook_views_api(). - */ -function advancedqueue_views_api() { - return array( - 'api' => 2, - 'path' => drupal_get_path('module', 'advancedqueue') . '/views', - ); -} - -/** * Implements hook_advancedqueue_entity_info(). */ function advancedqueue_entity_info() { @@ -50,3 +40,153 @@ function advancedqueue_entity_info() { ); return $entity_info; } + +/** + * Implements hook_cron(). + */ +function advancedqueue_cron() { + if (!variable_get('advancedqueue_use_cron', TRUE)) { + return; + } + + if (!$queues = advancedqueue_get_queues_info()) { + return; + } + + // @todo: Add variable to set interval? + $end = time() + 60; + foreach ($queues as $queue_name => $queue_info) { + $queue = DrupalQueue::get($queue_name); + + while ($item = $queue->claimItem()) { + if (time() > $end) { + // We've reached max execution time. + return; + } + advancedqueue_process_item($queue, $queue_name, $queue_info, $item, $end); + } + } +} + +/** + * Implements hook_form_FORM_ID_alter(). + * + * Add Advanced queue setting to use cron, to the cron settings page. + */ +function advancedqueue_form_system_cron_settings_alter(&$form, $form_state, $form_id) { + $form['advancedqueue_use_cron'] = array( + '#type' => 'checkbox', + '#title' => t('Process Advanced Queue via Cron'), + '#multiple' => TRUE, + '#description' => t('Enable to allow queue items to to be processed using Cron. This is a "poor man\'s" option that allows processing the queue, as the better solution would be to execute the Drush command via the command line.'), + '#default_value' => variable_get('advancedqueue_use_cron', TRUE), + ); +} + +/** + * Return queue(s) info. + * + * @params $queue_names + * Optional; Array with the queue names. If empty, return all the queues. + */ +function advancedqueue_get_queues_info($queue_names = array()) { + $queues_info = &drupal_static(__FUNCTION__, array()); + + if (empty($queues_info)) { + $queues_info = module_invoke_all('advanced_queue_info'); + + // Add default values. + foreach ($queues_info as &$queue_info) { + $queue_info += array( + 'delete when completed' => TRUE, + ); + } + drupal_alter('advanced_queue_info', $queues_info); + uasort($queues_info, 'drupal_sort_weight'); + } + + if ($queue_names) { + return array_intersect_key($queues_info, $queue_names); + } + + return $queues_info; +} + + +/** + * Process a queue item. + * + * @param $queue + * The queue object. + * @param $queue_name + * The queue name. + * @param $queue_info + * The queue info. + * @param $item + * The item to process. + * @param $end_time + * (Optional) The time the process should end. + */ +function advancedqueue_process_item($queue, $queue_name, $queue_info, $item, $end_time = FALSE) { + $function = $queue_info['worker callback']; + $params = array( + '@queue' => $queue_name, + '@id' => $item->item_id, + '@title' => !empty($item->title) ? $item->title : 'untitled', + ); + + advancedqueue_log_message(format_string('[@queue:@id] Starting processing item @title.', $params)); + + try { + // Pass the claimed item itself and end date along to the worker + // callback. + $output = $function($item, $end_time); + if (is_array($output)) { + $item->status = $output['status']; + $item->result = $output['result']; + } + else { + $item->status = $output ? ADVANCEDQUEUE_STATUS_SUCCESS : ADVANCEDQUEUE_STATUS_FAILURE; + } + } + catch (Exception $e) { + $item->status = ADVANCEDQUEUE_STATUS_FAILURE; + $params['!message'] = (string) $e; + advancedqueue_log_message(format_string('[!queue:!id] failed processing: !message', $params)); + } + + $params['@status'] = $item->status; + advancedqueue_log_message(format_string('[@queue:@id] Processing ended with result @status.', $params)); + + if ($queue_info['delete when completed']) { + // Item was processed, so we can "delete" it. This is not removing the + // item from the database, but rather updates it with the status. + $queue->deleteItem($item); + } +} + +/** + * Helper function to log message. In CLI we use Drush Log, otherwise watchdog. + * + * @param $message + * The message to log. + */ +function advancedqueue_log_message($message) { + if (drupal_is_cli()) { + drush_log($message); + } + else { + watchdog('advancedqueue', $message); + } +} + +/** + * Implements hook_views_api(). + */ +function advancedqueue_views_api() { + return array( + 'api' => 2, + 'path' => drupal_get_path('module', 'advancedqueue') . '/views', + ); +} + diff --git a/advancedqueue_example/advancedqueue_example.module b/advancedqueue_example/advancedqueue_example.module index 3a29798..e99b8cd 100644 --- a/advancedqueue_example/advancedqueue_example.module +++ b/advancedqueue_example/advancedqueue_example.module @@ -10,6 +10,7 @@ */ function advancedqueue_example_advanced_queue_info() { $items['example_queue'] = array( + 'label' => t('Example queue'), 'worker callback' => 'advancedqueue_example_worker', ); return $items; @@ -34,7 +35,7 @@ function advancedqueue_example_worker($item, $end_time = FALSE) { '@uid' => $data['uid'], '@time' => date('r', $data['timestamp']), ); - drush_print(dt('The "worker" is now processing a example task number @id for user ID @uid created at @time.', $params)); + advancedqueue_log_message(format_string('The "worker" is now processing a example task number @id for user ID @uid created at @time.', $params)); // For example purposes we will return an array with detailed message // for odd item IDs, and boolean for even ones. diff --git a/drush/advancedqueue.drush.inc b/drush/advancedqueue.drush.inc index d7413ee..07b0ce0 100644 --- a/drush/advancedqueue.drush.inc +++ b/drush/advancedqueue.drush.inc @@ -26,9 +26,9 @@ function advancedqueue_drush_command() { function drush_advancedqueue_process_queue($queue = NULL) { // Load information about the registred queues, and sort them by weight. - $all_queue_info = module_invoke_all('advanced_queue_info'); - drupal_alter('advanced_queue_info', $all_queue_info); - uasort($all_queue_info, 'drupal_sort_weight'); + if (!$queues_info = advancedqueue_get_queues_info()) { + return drush_set_error(dt('No queues exist.')); + } $all_option = drush_get_option('all'); if (!$all_option && empty($queue)) { @@ -36,20 +36,15 @@ function drush_advancedqueue_process_queue($queue = NULL) { } if ($all_option) { - $queues = $all_queue_info; + $queues = $queues_info; } else { // Validate queues. $queues = drupal_map_assoc(explode(',', $queue)); - $invalid_queues = array_diff_key($queues, $all_queue_info); - if ($invalid_queues) { + if ($invalid_queues = array_diff_key($queues, $queues_info)) { return drush_set_error(dt('The following queues are invalid: !queues. Aborting.', array('!queues' => implode(', ', $invalid_queues)))); } - $queues = array_intersect_key($all_queue_info, $queues); - } - - if (!$queues) { - return drush_set_error(dt('No queues exist.')); + $queues = array_intersect_key($queues_info, $queues); } // Run the worker for a certain period of time before killing it. @@ -68,7 +63,7 @@ function drush_advancedqueue_process_queue($queue = NULL) { $queue = DrupalQueue::get($queue_name); if ($item = $queue->claimItem()) { - drush_advancedqueue_process_item($queue, $queue_name, $queue_info, $item, $end); + advancedqueue_process_item($queue, $queue_name, $queue_info, $item, $end); continue 2; } } @@ -79,54 +74,3 @@ function drush_advancedqueue_process_queue($queue = NULL) { drush_log(dt('Timeout: exiting processing loop.')); } - -/** - * Process a queue item. - * - * @param $queue - * The queue object. - * @param $queue_name - * The queue name. - * @param $queue_info - * The queue info. - * @param $item - * The item to process. - * @param $end_time - * (Optional) The time the process should end. - */ -function drush_advancedqueue_process_item($queue, $queue_name, $queue_info, $item, $end_time = FALSE) { - $function = $queue_info['worker callback']; - $params = array( - '@queue' => $queue_name, - '@id' => $item->item_id, - '@title' => !empty($item->title) ? $item->title : dt('untitled'), - ); - drush_log(dt('[@queue:@id] Starting processing item @title.', $params)); - - try { - // Pass the claimed item itself and end date along to the worker - // callback. - $output = $function($item, $end_time); - if (is_array($output)) { - $item->status = $output['status']; - $item->result = $output['result']; - } - else { - $item->status = $output ? ADVANCEDQUEUE_STATUS_SUCCESS : ADVANCEDQUEUE_STATUS_FAILURE; - } - } - catch (Exception $e) { - $item->status = ADVANCEDQUEUE_STATUS_FAILURE; - $params['!message'] = (string) $e; - drush_log(dt('[!queue:!id] failed processing: !message', $params)); - } - - $params['@status'] = $item->status; - drush_log(dt('[@queue:@id] Processing ended with result @status.', $params)); - - if ($queue_info['delete when completed']) { - // Item was processed, so we can "delete" it. This is not removing the - // item from the database, but rather updates it with the status. - $queue->deleteItem($item); - } -} diff --git a/tests/advancedqueue.test b/tests/advancedqueue.test index 119d305..2cd74d3 100644 --- a/tests/advancedqueue.test +++ b/tests/advancedqueue.test @@ -1,6 +1,13 @@ 'Cron functionality', + 'description' => 'Workers called by cron, and admin options.', + 'group' => 'Advanced Queue', + ); + } + + function setUp() { + parent::setup('advancedqueue_test'); + variable_set('queue_default_class', 'AdvancedQueue'); + + // Admin user for admin page check. + $this->admin_user = $this->drupalCreateUser(array( + 'administer site configuration', + )); + } + + /** + * Test cron calls workers. + */ + function testCronWorker() { + $data = advancedqueue_test_populate_two_queues(); + $this->cronRun(); + + $count = 0; + // Check to see that all items were sent to the worker. + foreach ($data as $item) { + $count += advancedqueue_test_count_watchdog( + 'advancedqueue_test', + 'advancedqueue_test_worker(@item)', + array('@item' => print_r($item, TRUE)), + WATCHDOG_INFO + ); + } + + $this->assertEqual($count, 8, t('All queued items sent to worker.')); + } + + /** + * Test admin page options. + */ + function testAdminOption() { + $this->drupalLogin($this->admin_user); + + $this->drupalGet('admin/config/system/cron'); + $this->assertField('edit-advancedqueue-use-cron', t('Advancedqueue use cron option.')); + $this->assertFieldChecked('edit-advancedqueue-use-cron', t('Advancedqueue use cron option default checked.')); + $edit = array( + 'advancedqueue_use_cron' => FALSE, + ); + $this->drupalPost(NULL, $edit, t('Save configuration')); + $this->assertNoFieldChecked('edit-advancedqueue-use-cron', t('Advancedqueue cron disabled successfully.')); + + $this->drupalLogout(); + } +} diff --git a/tests/advancedqueue_test.inc b/tests/advancedqueue_test.inc new file mode 100644 index 0000000..5348328 --- /dev/null +++ b/tests/advancedqueue_test.inc @@ -0,0 +1,73 @@ +createQueue(); + $queue2 = DrupalQueue::get('advancedqueue_test_2'); + $queue2->createQueue(); + + // Create four items in queue 1. + $data = array(); + for ($i = 0; $i < 4; $i++) { + $data[] = $item = array(DrupalTestCase::randomName() => DrupalTestCase::randomName()); + $queue1->createItem($item); + } + + // Create four items in queue 2. + for ($i = 0; $i < 4; $i++) { + $data[] = $item = array(DrupalTestCase::randomName() => DrupalTestCase::randomName()); + $queue2->createItem($item); + } + + return $data; +} + +/** + * Verify log entries exist. + * + * Called in the same way of the expected original watchdog() execution. Based + * on ModuleTestCase::assertLogMessage(). + * + * @param string $type + * The category to which this message belongs. + * @param string $message + * The message to store in the log. Keep $message translatable + * by not concatenating dynamic values into it! Variables in the + * message should be added by using placeholder strings alongside + * the variables argument to declare the value of the placeholders. + * See t() for documentation on how $message and $variables interact. + * @param array $variables + * Array of variables to replace in the message on display or + * NULL if message is already translated or not possible to + * translate. + * @param int $severity + * The severity of the message, as per RFC 3164. + * @param string $link + * A link to associate with the message. + */ +function advancedqueue_test_count_watchdog($type, $message, $variables = array(), $severity = WATCHDOG_NOTICE, $link = '') { + $count = db_select('watchdog', 'w') + ->condition('type', $type) + ->condition('message', $message) + ->condition('variables', serialize($variables)) + ->condition('severity', $severity) + ->condition('link', $link) + ->countQuery() + ->execute() + ->fetchField(); + return $count; +} diff --git a/tests/advancedqueue_test.info b/tests/advancedqueue_test.info new file mode 100644 index 0000000..c09cd96 --- /dev/null +++ b/tests/advancedqueue_test.info @@ -0,0 +1,9 @@ +core = 7.x +name = Advanced Queues Test +description = Simpletests for advancedqueue. +package = Other + +dependencies[] = advancedqueue +dependencies[] = simpletest + +files[] = advancedqueue.test diff --git a/tests/advancedqueue_test.module b/tests/advancedqueue_test.module new file mode 100644 index 0000000..1456ef7 --- /dev/null +++ b/tests/advancedqueue_test.module @@ -0,0 +1,33 @@ + 'advancedqueue_test_worker', + ); + $queues['advancedqueue_test_2'] = array( + 'worker callback' => 'advancedqueue_test_worker', + ); + return $queues; +} + +/** + * Worker callback. + */ +function advancedqueue_test_worker($item) { + watchdog( + 'advancedqueue_test', + 'advancedqueue_test_worker(@item)', + array('@item' => print_r($item->data, TRUE)), + WATCHDOG_INFO + ); + + return TRUE; +}