/**
  * FIXME: Singleton pattern should be removed when dependency-injection
  * becomes available.
  *
  * @param $forceNew bool
  */
 static function &singleton($forceNew = FALSE)
 {
     if ($forceNew || !self::$_singleton) {
         self::$_singleton = new CRM_Queue_Service();
     }
     return self::$_singleton;
 }

 static function getRunner($skipEndUrl = FALSE)
 {
     // Setup the Queue
     $queue = CRM_Queue_Service::singleton()->create(array('name' => self::QUEUE_NAME, 'type' => 'Sql', 'reset' => TRUE));
     // reset pull stats.
     CRM_Core_BAO_Setting::setItem(array(), CRM_Mailchimp_Form_Setting::MC_SETTING_GROUP, 'pull_stats');
     $stats = array();
     // We need to process one list at a time.
     $groups = CRM_Mailchimp_Utils::getGroupsToSync(array(), null, $membership_only = TRUE);
     if (!$groups) {
         // Nothing to do.
         return FALSE;
     }
     // Each list is a task.
     $listCount = 1;
     foreach ($groups as $group_id => $details) {
         $stats[$details['list_id']] = array('mc_count' => 0, 'c_count' => 0, 'in_sync' => 0, 'added' => 0, 'removed' => 0);
         $identifier = "List " . $listCount++ . " " . $details['civigroup_title'];
         $task = new CRM_Queue_Task(array('CRM_Mailchimp_Form_Pull', 'syncPullList'), array($details['list_id'], $identifier), "Preparing queue for {$identifier}");
         // Add the Task to the Queue
         $queue->createItem($task);
     }
     // Setup the Runner
     $runnerParams = array('title' => ts('Import From Mailchimp'), 'queue' => $queue, 'errorMode' => CRM_Queue_Runner::ERROR_ABORT, 'onEndUrl' => CRM_Utils_System::url(self::END_URL, self::END_PARAMS, TRUE, NULL, FALSE));
     // Skip End URL to prevent redirect
     // if calling from cron job
     if ($skipEndUrl == TRUE) {
         unset($runnerParams['onEndUrl']);
     }
     $runner = new CRM_Queue_Runner($runnerParams);
     static::updatePullStats($stats);
     return $runner;
 }

 public function setUp()
 {
     parent::setUp();
     $this->queueService = CRM_Queue_Service::singleton(TRUE);
     $this->queue = $this->queueService->create(array('type' => 'Sql', 'name' => 'test-queue'));
     self::$_recordedValues = array();
 }

 function run()
 {
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => self::QUEUE_NAME, 'reset' => TRUE));
     for ($i = 0; $i < 5; $i++) {
         $queue->createItem(new CRM_Queue_Task(array('CRM_Demoqueue_Page_DemoQueue', 'doMyWork'), array($i, "Task {$i} takes {$i} second(s)"), "Task {$i}"));
         if ($i == 2) {
             $queue->createItem(new CRM_Queue_Task(array('CRM_Demoqueue_Page_DemoQueue', 'addMoreWork'), array(), "Add More Work"));
         }
     }
     $runner = new CRM_Queue_Runner(array('title' => ts('Demo Queue Runner'), 'queue' => $queue, 'onEnd' => array('CRM_Demoqueue_Page_DemoQueue', 'onEnd'), 'onEndUrl' => CRM_Utils_System::url('civicrm/demo-queue/done')));
     $runner->runAllViaWeb();
     // does not return
 }

 /**
  * Build a queue of tasks by dividing dupe pairs in batches.
  */
 public static function getRunner()
 {
     $rgid = CRM_Utils_Request::retrieve('rgid', 'Positive', $this, FALSE, 0);
     $gid = CRM_Utils_Request::retrieve('gid', 'Positive', $this, FALSE, 0);
     $action = CRM_Utils_Request::retrieve('action', 'String', CRM_Core_DAO::$_nullObject);
     $mode = CRM_Utils_Request::retrieve('mode', 'String', CRM_Core_DAO::$_nullObject, FALSE, 'safe');
     $contactType = CRM_Core_DAO::getFieldValue('CRM_Dedupe_DAO_RuleGroup', $rgid, 'contact_type');
     $cacheKeyString = "merge {$contactType}";
     $cacheKeyString .= $rgid ? "_{$rgid}" : '_0';
     $cacheKeyString .= $gid ? "_{$gid}" : '_0';
     $urlQry = "reset=1&action=update&rgid={$rgid}";
     $urlQry = $gid ? $urlQry . "&gid={$gid}" : $urlQry;
     if ($mode == 'aggressive' && !CRM_Core_Permission::check('force merge duplicate contacts')) {
         CRM_Core_Session::setStatus(ts('You do not have permission to force merge duplicate contact records'), ts('Permission Denied'), 'error');
         CRM_Utils_System::redirect(CRM_Utils_System::url('civicrm/contact/dedupefind', $urlQry));
     }
     // Setup the Queue
     $queue = CRM_Queue_Service::singleton()->create(array('name' => $cacheKeyString, 'type' => 'Sql', 'reset' => TRUE));
     $where = NULL;
     if ($action == CRM_Core_Action::MAP) {
         $where = "pn.is_selected = 1";
         $isSelected = 1;
     } else {
         // else merge all (2)
         $isSelected = 2;
     }
     $total = CRM_Core_BAO_PrevNextCache::getCount($cacheKeyString, NULL, $where);
     if ($total <= 0) {
         // Nothing to do.
         CRM_Utils_System::redirect(CRM_Utils_System::url('civicrm/contact/dedupefind', $urlQry));
     }
     // reset merge stats, so we compute new stats
     CRM_Dedupe_Merger::resetMergeStats($cacheKeyString);
     for ($i = 1; $i <= ceil($total / self::BATCHLIMIT); $i++) {
         $task = new CRM_Queue_Task(array('CRM_Contact_Page_DedupeMerge', 'callBatchMerge'), array($rgid, $gid, $mode, TRUE, self::BATCHLIMIT, $isSelected), "Processed " . $i * self::BATCHLIMIT . " pair of duplicates out of " . $total);
         // Add the Task to the Queue
         $queue->createItem($task);
     }
     // Setup the Runner
     $urlQry .= "&context=conflicts";
     $runner = new CRM_Queue_Runner(array('title' => ts('Merging Duplicates..'), 'queue' => $queue, 'errorMode' => CRM_Queue_Runner::ERROR_ABORT, 'onEndUrl' => CRM_Utils_System::url('civicrm/contact/dedupefind', $urlQry, TRUE, NULL, FALSE)));
     return $runner;
 }

function civicrm_api3_civi_mailchimp_sync($params)
{
    CRM_CiviMailchimp_BAO_SyncLog::deleteOldMessages();
    $records_to_process = $params['records_to_process_per_run'];
    $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => 'mailchimp-sync', 'reset' => FALSE));
    if ($queue->numberOfItems() > 0) {
        $runner = new CRM_Queue_Runner(array('title' => ts('Sync Contacts to Mailchimp'), 'queue' => $queue));
        $continue_to_next_item = TRUE;
        $records_processed = 0;
        while ($continue_to_next_item && $records_processed < $records_to_process) {
            $record = $runner->runNext();
            if ($record['is_error']) {
                // Get the current Queue Item being worked on to allow for better error
                // reporting and logging.
                $query = "\n          SELECT\n            id,\n            data\n          FROM\n            civicrm_queue_item\n          WHERE\n            queue_name = 'mailchimp-sync'\n          ORDER BY\n            weight ASC,\n            id ASC\n          LIMIT 1\n        ";
                $item = CRM_Core_DAO::executeQuery($query);
                while ($item->fetch()) {
                    $item_data = unserialize($item->data);
                    $message = "[{$item_data->arguments[0]}] There was an error syncing contacts to Mailchimp.";
                    $exception_name = '';
                    if (!empty($record['exception'])) {
                        $exception_name = get_class($record['exception']);
                        $message = "[{$item_data->arguments[0]}] {$exception_name}: {$record['exception']->getMessage()}.";
                    }
                    $message .= " Mailchimp List ID: {$item_data->arguments[1]}. {$records_processed} records were successfully synced before this error.";
                    $error = array('code' => $exception_name, 'message' => $message, 'exception' => $record['exception']);
                    CRM_Core_Error::debug_var('Fatal Error Details', $error);
                    CRM_Core_Error::backtrace('backTrace', TRUE);
                    CRM_CiviMailchimp_BAO_SyncLog::saveMessage('error', 'civicrm_to_mailchimp', $message, $item_data, $item->id);
                    return civicrm_api3_create_error($message);
                }
            }
            $continue_to_next_item = $record['is_continue'];
            $records_processed++;
        }
        $message = ts('%1 records were successfully synced to Mailchimp.', array(1 => $records_processed));
        CRM_CiviMailchimp_BAO_SyncLog::saveMessage('success', 'civicrm_to_mailchimp', $message);
        return civicrm_api3_create_success($records_processed);
    }
}

 public static function getRunner($skipEndUrl = FALSE, $dry_run = FALSE)
 {
     // Setup the Queue
     $queue = CRM_Queue_Service::singleton()->create(array('name' => self::QUEUE_NAME, 'type' => 'Sql', 'reset' => TRUE));
     // reset pull stats.
     $stats = ['dry_run' => $dry_run];
     CRM_Core_BAO_Setting::setItem($stats, CRM_Mailchimp_Form_Setting::MC_SETTING_GROUP, 'pull_stats');
     // We need to process one list at a time.
     $groups = CRM_Mailchimp_Utils::getGroupsToSync(array(), null, $membership_only = TRUE);
     if (!$groups) {
         // Nothing to do.
         return FALSE;
     }
     // Each list is a task.
     $listCount = 1;
     foreach ($groups as $group_id => $details) {
         if (empty($details['list_name'])) {
             // This list has been deleted at Mailchimp, or for some other reason we
             // could not access its name. Best not to sync it.
             continue;
         }
         $stats[$details['list_id']] = array('mc_count' => 0, 'c_count' => 0, 'in_sync' => 0, 'added' => 0, 'removed' => 0);
         $identifier = "List " . $listCount++ . " " . $details['civigroup_title'];
         $task = new CRM_Queue_Task(array('CRM_Mailchimp_Form_Pull', 'syncPullList'), array($details['list_id'], $identifier, $dry_run), "{$identifier}: collecting data from CiviCRM...");
         // Add the Task to the Queue
         $queue->createItem($task);
     }
     // Setup the Runner
     $runnerParams = array('title' => ($dry_run ? ts('Dry Run: ') : '') . ts('Mailchimp Pull Sync: update CiviCRM from Mailchimp'), 'queue' => $queue, 'errorMode' => CRM_Queue_Runner::ERROR_ABORT, 'onEndUrl' => CRM_Utils_System::url(self::END_URL, self::END_PARAMS, TRUE, NULL, FALSE));
     // Skip End URL to prevent redirect
     // if calling from cron job
     if ($skipEndUrl == TRUE) {
         unset($runnerParams['onEndUrl']);
     }
     $runner = new CRM_Queue_Runner($runnerParams);
     static::updatePullStats($stats);
     return $runner;
 }

 function postProcess()
 {
     $current_status_ids = array();
     $dao = CRM_Core_DAO::executeQuery("SELECT id from civicrm_membership_status where is_current_member = 1");
     while ($dao->fetch()) {
         $current_status_ids[] = $dao->id;
     }
     $formValues = $this->exportValues();
     if (!isset($formValues['member_status_id'])) {
         $formValues['member_status_id'] = array();
     }
     $birth_date_from = CRM_Utils_Date::processDate($formValues['birth_date_from']);
     $birth_date_to = CRM_Utils_Date::processDate($formValues['birth_date_to']);
     $selector = new CRM_Rood_UpgradeSelector();
     $original_where = $selector->getWhere();
     $selector->setData($formValues['rood_mtype'], array_keys($formValues['member_status_id']), $birth_date_from, $birth_date_to);
     $selector->store();
     $where = $selector->getWhere();
     $count = CRM_Core_DAO::singleValueQuery("SELECT COUNT(*)\n      FROM civicrm_membership\n      INNER JOIN civicrm_contact ON civicrm_membership.contact_id = civicrm_contact.id\n      LEFT JOIN civicrm_membership sp_membership ON civicrm_contact.id = sp_membership.contact_id AND sp_membership.membership_type_id = '{$formValues['sp_mtype']}' AND sp_membership.status_id IN (" . implode(', ', $current_status_ids) . ")\n      " . $where . " AND sp_membership.id is null");
     $this->assign('found', $count);
     if ($where == $original_where && isset($_POST['continue']) && !empty($_POST['continue'])) {
         $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => 'nl.sp.rood.upgrade', 'reset' => TRUE));
         $dao = CRM_Core_DAO::executeQuery("SELECT civicrm_membership.id\n       FROM civicrm_membership\n      INNER JOIN civicrm_contact ON civicrm_membership.contact_id = civicrm_contact.id\n      LEFT JOIN civicrm_membership sp_membership ON civicrm_contact.id = sp_membership.contact_id AND sp_membership.membership_type_id = '{$formValues['sp_mtype']}' AND sp_membership.status_id IN (" . implode(', ', $current_status_ids) . ")\n      " . $where . " AND sp_membership.id is null\n      ");
         $i = 1;
         while ($dao->fetch()) {
             $title = ts('Upgrade Rood lidmaatschappen %1 van %2', array(1 => $i, 2 => $count));
             //create a task without parameters
             $task = new CRM_Queue_Task(array('CRM_Rood_MembershipUpgrade', 'UpgradeFromQueue'), array($dao->id, $formValues), $title);
             //now add this task to the queue
             $queue->createItem($task);
             $i++;
         }
         $runner = new CRM_Queue_Runner(array('title' => ts('Upgrade rood lidmaatschappen'), 'queue' => $queue, 'errorMode' => CRM_Queue_Runner::ERROR_ABORT, 'onEnd' => array('CRM_Rood_Form_UpgradeRoodMembership', 'onEnd'), 'onEndUrl' => CRM_Utils_System::url('civicrm/member/upgrade_rood', 'reset=1')));
         $runner->runAllViaWeb();
         // does not return
     }
 }

 function testClearQueueItem()
 {
     $mailchimp_list_id = 'MailchimpListsTestListA';
     $mailchimp_sync_setting = CRM_CiviMailchimp_BAO_SyncSettingsTest::createTestGroupAndSyncSettings('Test Group testClearQueueItem', $mailchimp_list_id);
     $group = CRM_CiviMailchimp_Utils::getGroupById($mailchimp_sync_setting->civicrm_group_id);
     $params = CRM_CiviMailchimp_UtilsTest::sampleContactParams();
     $contact = CRM_Contact_BAO_Contact::create($params);
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => 'mailchimp-sync', 'reset' => TRUE));
     civimailchimp_civicrm_contact_removed_from_group($group, $contact);
     $item = $queue->claimItem($lease_time = 0);
     $existing_mailchimp_sync_log = self::createTestLogMessage('This is a test error message', $details = NULL, $direction = 'civicrm_to_mailchimp', $type = 'error', $item->id);
     CRM_CiviMailchimp_BAO_SyncLog::clearQueueItem($item->id);
     $query = "\n      SELECT\n        *\n      FROM\n        civicrm_queue_item\n      WHERE\n        queue_name = 'mailchimp-sync'\n      AND\n        id = %1\n      LIMIT 1\n      ";
     $params = array(1 => array($item->id, 'Integer'));
     $new_item = CRM_Core_DAO::executeQuery($query, $params, TRUE, 'CRM_Queue_DAO_QueueItem');
     $mailchimp_sync_log = CRM_CiviMailchimp_BAO_SyncLog::findById($existing_mailchimp_sync_log->id);
     $this->assertFalse($new_item->fetch());
     $this->assertNull($mailchimp_sync_log->civicrm_queue_item_id);
 }

 /**
  * Syntactic sugar for adding a task which (a) is in this class and (b) has
  * a high priority.
  *
  * After passing the $funcName, you can also pass parameters that will go to
  * the function. Note that all params must be serializable.
  */
 protected function addTask($title, $funcName)
 {
     $queue = CRM_Queue_Service::singleton()->load(array('type' => 'Sql', 'name' => CRM_Upgrade_Form::QUEUE_NAME));
     $args = func_get_args();
     $title = array_shift($args);
     $funcName = array_shift($args);
     $task = new CRM_Queue_Task(array(get_class($this), $funcName), $args, $title);
     $queue->createItem($task, array('weight' => -1));
 }

 /**
  * Fill a queue with upgrade tasks.
  *
  * @return CRM_Queue_Queue
  */
 public static function createQueue()
 {
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => self::QUEUE_NAME, 'reset' => TRUE));
     CRM_Utils_Hook::upgrade('enqueue', $queue);
     return $queue;
 }

 function test_civicrm_api3_civi_mailchimp_sync()
 {
     $action = 'subscribeContactToMailchimpList';
     $mailchimp_list_id = 'MailchimpListsTestListA';
     $mailchimp_interest_groups = array('MailchimpTestInterestGroupingA_MailchimpTestInterestGroupA', 'MailchimpTestInterestGroupingA_MailchimpTestInterestGroupC');
     $mailchimp_sync_setting = CRM_CiviMailchimp_BAO_SyncSettingsTest::createTestGroupAndSyncSettings('Test Group test_civicrm_api3_civi_mailchimp_sync', $mailchimp_list_id, $mailchimp_interest_groups);
     $merge_fields = CRM_CiviMailchimp_Utils::getMailchimpMergeFields();
     $params = CRM_CiviMailchimp_UtilsTest::sampleContactParams();
     $contact = CRM_Contact_BAO_Contact::create($params);
     $merge_vars = CRM_CiviMailchimp_Utils::formatMailchimpMergeVars($merge_fields, $contact);
     CRM_CiviMailchimp_Utils::addMailchimpSyncQueueItem($action, $mailchimp_list_id, $params['email'][0]['email'], $merge_vars);
     $action = 'unsubscribeContactFromMailchimpList';
     CRM_CiviMailchimp_Utils::addMailchimpSyncQueueItem($action, $mailchimp_list_id, $params['email'][0]['email'], $merge_vars);
     $job_params['records_to_process_per_run'] = 100;
     civicrm_api3_civi_mailchimp_sync($job_params);
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => 'mailchimp-sync', 'reset' => FALSE));
     $this->assertEquals(0, $queue->numberOfItems());
 }

 /**
  * Per-provider tests
  */
 public function setUp()
 {
     parent::setUp();
     $this->queueService = CRM_Queue_Service::singleton(TRUE);
 }

 static function getRunner($skipEndUrl = FALSE, $fromContactId, $toContactId, $batchSize)
 {
     // Setup the Queue
     $queue = CRM_Queue_Service::singleton()->create(array('name' => self::QUEUE_NAME, 'type' => 'Sql', 'reset' => TRUE));
     CRM_Utils_Normalize::setSetting(array('contact' => 0, 'phone' => 0, 'address' => 0), 'normalization_stats');
     for ($startId = $fromContactId; $startId <= $toContactId; $startId += $batchSize) {
         $endId = $startId + $batchSize - 1;
         $title = ts('Normalizing contacts (%1 => %2)', array(1 => $startId, 2 => $endId));
         $task = new CRM_Queue_Task(array('CRM_Admin_Form_Setting_Normalize', 'normalizeContacts'), array($startId, $endId, $title), "Preparing queue for {$title}");
         // Add the Task to the Queue
         $queue->createItem($task);
     }
     // Setup the Runner
     $runnerParams = array('title' => ts('Contact Normalization'), 'queue' => $queue, 'errorMode' => CRM_Queue_Runner::ERROR_ABORT, 'onEndUrl' => CRM_Utils_System::url(self::END_URL, self::END_PARAMS, TRUE, NULL, FALSE));
     // Skip End URL to prevent redirect
     // if calling from cron job
     if ($skipEndUrl == TRUE) {
         unset($runnerParams['onEndUrl']);
     }
     $runner = new CRM_Queue_Runner($runnerParams);
     return $runner;
 }

 /**
  * Add a mailchimp sync item to the queue.
  */
 static function addMailchimpSyncQueueItem($action, $mailchimp_list_id, $email, $merge_vars = array())
 {
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => 'mailchimp-sync', 'reset' => FALSE));
     $queue->createItem(new CRM_Queue_Task(array('CRM_CiviMailchimp_Utils', 'processCiviMailchimpQueueItem'), array($action, $mailchimp_list_id, $email, $merge_vars)));
 }

 private function __construct()
 {
     $this->queue = CRM_Queue_Service::singleton()->create(array('type' => 'Sql', 'name' => self::QUEUE_NAME, 'reset' => false));
 }

 /**
  * Save an action into a queue for delayed processing
  *
  * @param \DateTime $delayTo
  * @param \CRM_Civirules_Action $action
  * @param \CRM_Civirules_TriggerData_TriggerData $triggerData
  */
 protected static function delayAction(DateTime $delayTo, CRM_Civirules_Action $action, CRM_Civirules_TriggerData_TriggerData $triggerData)
 {
     $queue = CRM_Queue_Service::singleton()->create(array('type' => 'Civirules', 'name' => self::QUEUE_NAME, 'reset' => false));
     //create a task with the action and eventData as parameters
     $task = new CRM_Queue_Task(array('CRM_Civirules_Engine', 'executeDelayedAction'), array($action, $triggerData));
     //save the task with a delay
     $dao = new CRM_Queue_DAO_QueueItem();
     $dao->queue_name = $queue->getName();
     $dao->submit_time = CRM_Utils_Time::getTime('YmdHis');
     $dao->data = serialize($task);
     $dao->weight = 0;
     //weight, normal priority
     $dao->release_time = $delayTo->format('YmdHis');
     $dao->save();
 }

 /**
  * Fill the queue with upgrade tasks.
  *
  * @param string $currentVer
  *   the original revision.
  * @param string $latestVer
  *   the target (final) revision.
  * @param string $postUpgradeMessageFile
  *   path of a modifiable file which lists the post-upgrade messages.
  *
  * @return CRM_Queue_Service
  */
 public static function buildQueue($currentVer, $latestVer, $postUpgradeMessageFile)
 {
     $upgrade = new CRM_Upgrade_Form();
     // hack to make 4.0.x (D7,J1.6) codebase go through 3.4.x (d6, J1.5) upgrade files,
     // since schema wise they are same
     if (CRM_Upgrade_Form::getRevisionPart($currentVer) == '4.0') {
         $currentVer = str_replace('4.0.', '3.4.', $currentVer);
     }
     // Ensure that queue can be created
     if (!CRM_Queue_BAO_QueueItem::findCreateTable()) {
         CRM_Core_Error::fatal(ts('Failed to find or create queueing table'));
     }
     $queue = CRM_Queue_Service::singleton()->create(array('name' => self::QUEUE_NAME, 'type' => 'Sql', 'reset' => TRUE));
     $revisions = $upgrade->getRevisionSequence();
     foreach ($revisions as $rev) {
         // proceed only if $currentVer < $rev
         if (version_compare($currentVer, $rev) < 0) {
             $beginTask = new CRM_Queue_Task(array('CRM_Upgrade_Form', 'doIncrementalUpgradeStart'), array($rev), "Begin Upgrade to {$rev}");
             $queue->createItem($beginTask);
             $task = new CRM_Queue_Task(array('CRM_Upgrade_Form', 'doIncrementalUpgradeStep'), array($rev, $currentVer, $latestVer, $postUpgradeMessageFile), "Upgrade DB to {$rev}");
             $queue->createItem($task);
             $task = new CRM_Queue_Task(array('CRM_Upgrade_Form', 'doIncrementalUpgradeFinish'), array($rev, $currentVer, $latestVer, $postUpgradeMessageFile), "Finish Upgrade DB to {$rev}");
             $queue->createItem($task);
         }
     }
     return $queue;
 }