/**
  * Register our queue handlers
  *
  * @param QueueManager $qm Current queue manager
  *
  * @return boolean hook value
  */
 function onEndInitializeQueueManager($qm)
 {
     $qm->connect('siterem', 'SiteConfirmReminderHandler');
     $qm->connect('uregrem', 'UserConfirmRegReminderHandler');
     $qm->connect('uinvrem', 'UserInviteReminderHandler');
     return true;
 }

 /**
  * Handler for POST content updates from the hub
  */
 function handlePost()
 {
     $feedid = $this->arg('feed');
     common_log(LOG_INFO, "POST for feed id {$feedid}");
     if (!$feedid) {
         // TRANS: Server exception thrown when referring to a non-existing or empty feed.
         throw new ServerException(_m('Empty or invalid feed id.'), 400);
     }
     $feedsub = FeedSub::getKV('id', $feedid);
     if (!$feedsub instanceof FeedSub) {
         // TRANS: Server exception. %s is a feed ID.
         throw new ServerException(sprintf(_m('Unknown PuSH feed id %s'), $feedid), 400);
     }
     $hmac = '';
     if (isset($_SERVER['HTTP_X_HUB_SIGNATURE'])) {
         $hmac = $_SERVER['HTTP_X_HUB_SIGNATURE'];
     }
     $post = file_get_contents('php://input');
     // Queue this to a background process; we should return
     // as quickly as possible from a distribution POST.
     // If queues are disabled this'll process immediately.
     $data = array('feedsub_id' => $feedsub->id, 'post' => $post, 'hmac' => $hmac);
     $qm = QueueManager::get();
     $qm->enqueue($data, 'pushin');
 }

 function queueBackup()
 {
     $cur = common_current_user();
     $qm = QueueManager::get();
     $qm->enqueue($cur->id, 'backoff');
     $this->showPage();
 }

 static function get()
 {
     if (empty(self::$qm)) {
         if (Event::handle('StartNewQueueManager', array(&self::$qm))) {
             $enabled = common_config('queue', 'enabled');
             $type = common_config('queue', 'subsystem');
             if (!$enabled) {
                 // does everything immediately
                 self::$qm = new UnQueueManager();
             } else {
                 switch ($type) {
                     case 'db':
                         self::$qm = new DBQueueManager();
                         break;
                     case 'stomp':
                         self::$qm = new StompQueueManager();
                         break;
                     default:
                         throw new ServerException("No queue manager class for type '{$type}'");
                 }
             }
         }
     }
     return self::$qm;
 }

 /**
  * Initialize IoManagers for the currently configured site
  * which are appropriate to this instance.
  */
 function initManagers()
 {
     if (common_config('xmpp', 'enabled')) {
         $qm = QueueManager::get();
         $qm->setActiveGroup('xmpp');
         $this->instantiate($qm);
         $this->instantiate(XmppManager::get());
         $this->instantiate($this->processManager);
     }
 }

function doSendControl($message, $event, $param = '')
{
    print $message;
    $qm = QueueManager::get();
    if ($qm->sendControlSignal($event, $param)) {
        print " sent.\n";
    } else {
        print " FAILED.\n";
    }
}

 /**
  * Initialize IoManagers for the currently configured site
  * which are appropriate to this instance.
  */
 function initManagers()
 {
     $classes = array();
     if (Event::handle('StartImDaemonIoManagers', array(&$classes))) {
         $qm = QueueManager::get();
         $qm->setActiveGroup('im');
         $classes[] = $qm;
         $classes[] = $this->processManager;
     }
     Event::handle('EndImDaemonIoManagers', array(&$classes));
     foreach ($classes as $class) {
         $this->instantiate($class);
     }
 }

 /**
  * Set up queue handlers for outgoing hub pushes
  * @param QueueManager $qm
  * @return boolean hook return
  */
 function onEndInitializeQueueManager(QueueManager $qm)
 {
     // Prepare outgoing distributions after notice save.
     $qm->connect('ostatus', 'OStatusQueueHandler');
     // Outgoing from our internal PuSH hub
     $qm->connect('hubconf', 'HubConfQueueHandler');
     $qm->connect('hubprep', 'HubPrepQueueHandler');
     $qm->connect('hubout', 'HubOutQueueHandler');
     // Outgoing Salmon replies (when we don't need a return value)
     $qm->connect('salmon', 'SalmonQueueHandler');
     // Incoming from a foreign PuSH hub
     $qm->connect('pushin', 'PushInQueueHandler');
     return true;
 }

 public function handle($user)
 {
     if (!$user instanceof User) {
         common_log(LOG_ERR, "Got a bogus user, not deleting");
         return true;
     }
     $user = User::getKV('id', $user->id);
     if (!$user) {
         common_log(LOG_INFO, "User {$user->nickname} was deleted before we got here.");
         return true;
     }
     try {
         if (!$user->hasRole(Profile_role::DELETED)) {
             common_log(LOG_INFO, "User {$user->nickname} is not pending deletion; aborting.");
             return true;
         }
     } catch (UserNoProfileException $unp) {
         common_log(LOG_INFO, "Deleting user {$user->nickname} with no profile... probably a good idea!");
     }
     $notice = $this->getNextBatch($user);
     if ($notice->N) {
         common_log(LOG_INFO, "Deleting next {$notice->N} notices by {$user->nickname}");
         while ($notice->fetch()) {
             $del = clone $notice;
             $del->delete();
         }
         // @todo improve reliability in case we died during the above deletions
         // with a fatal error. If the job is lost, we should perform some kind
         // of garbage collection later.
         // Queue up the next batch.
         $qm = QueueManager::get();
         $qm->enqueue($user, 'deluser');
     } else {
         // Out of notices? Let's finish deleting this profile!
         try {
             $user->getProfile()->delete();
         } catch (UserNoProfileException $e) {
             // in case a profile didn't exist for some reason, just delete the User directly
             $user->delete();
         }
         common_log(LOG_INFO, "User {$user->id} {$user->nickname} deleted.");
         return true;
     }
     return true;
 }

 /**
  * Handle the site
  *
  * @param array $remitem type of reminder to send and any special options
  * @return boolean true on success, false on failure
  */
 function handle($remitem)
 {
     list($type, $opts) = $remitem;
     $qm = QueueManager::get();
     try {
         switch ($type) {
             case UserConfirmRegReminderHandler::REGISTER_REMINDER:
                 $confirm = new Confirm_address();
                 $confirm->address_type = $type;
                 $confirm->find();
                 while ($confirm->fetch()) {
                     try {
                         $qm->enqueue(array($confirm, $opts), 'uregrem');
                     } catch (Exception $e) {
                         common_log(LOG_WARNING, $e->getMessage());
                         continue;
                     }
                 }
                 break;
             case UserInviteReminderHandler::INVITE_REMINDER:
                 $invitation = new Invitation();
                 // Only send one reminder (the latest one), regardless of how many invitations a user has
                 $sql = 'SELECT * FROM (SELECT * FROM invitation WHERE registered_user_id IS NULL ORDER BY created DESC) invitees GROUP BY invitees.address';
                 $invitation->query($sql);
                 while ($invitation->fetch()) {
                     try {
                         $qm->enqueue(array($invitation, $opts), 'uinvrem');
                     } catch (Exception $e) {
                         common_log(LOG_WARNING, $e->getMessage());
                         continue;
                     }
                 }
                 break;
             default:
                 // WTF?
                 common_log(LOG_ERR, "Received unknown confirmation address type", __FILE__);
         }
     } catch (Exception $e) {
         common_log(LOG_ERR, $e->getMessage());
         return false;
     }
     return true;
 }

 /**
  * Handle the site
  * 
  * @param mixed $object
  * @return boolean true on success, false on failure
  */
 function handle($object)
 {
     $qm = QueueManager::get();
     try {
         // Enqueue a summary for all users
         $user = new User();
         $user->find();
         while ($user->fetch()) {
             try {
                 $qm->enqueue($user->id, 'usersum');
             } catch (Exception $e) {
                 common_log(LOG_WARNING, $e->getMessage());
                 continue;
             }
         }
     } catch (Exception $e) {
         common_log(LOG_WARNING, $e->getMessage());
     }
     return true;
 }

 function handle($data)
 {
     list($user, $xml, $trusted) = $data;
     try {
         $doc = DOMDocument::loadXML($xml);
         $feed = $doc->documentElement;
         if ($feed->namespaceURI != Activity::ATOM || $feed->localName != 'feed') {
             // TRANS: Client exception thrown when an imported feed is not an Atom feed.
             throw new ClientException(_("Not an Atom feed."));
         }
         $author = ActivityUtils::getFeedAuthor($feed);
         if (empty($author)) {
             // TRANS: Client exception thrown when an imported feed does not have an author.
             throw new ClientException(_("No author in the feed."));
         }
         if (empty($user)) {
             if ($trusted) {
                 $user = $this->userFromAuthor($author);
             } else {
                 // TRANS: Client exception thrown when an imported feed does not have an author that
                 // TRANS: can be associated with a user.
                 throw new ClientException(_("Cannot import without a user."));
             }
         }
         $activities = $this->getActivities($feed);
         $qm = QueueManager::get();
         foreach ($activities as $activity) {
             $qm->enqueue(array($user, $author, $activity, $trusted), 'actimp');
         }
     } catch (ClientException $ce) {
         common_log(LOG_WARNING, $ce->getMessage());
         return true;
     } catch (ServerException $se) {
         common_log(LOG_ERR, $ce->getMessage());
         return false;
     } catch (Exception $e) {
         common_log(LOG_ERR, $ce->getMessage());
         return false;
     }
 }

 public static function enqueueNewFeeds(array $args = array())
 {
     if (!isset($args['interval']) || !is_int($args['interval']) || $args['interval'] <= 0) {
         $args['interval'] = self::DEFAULT_INTERVAL;
     }
     $args['interval'] *= 60;
     // minutes to seconds
     $feedsub = new FeedSub();
     $feedsub->sub_state = 'nohub';
     // Find feeds that haven't been polled within the desired interval,
     // though perhaps we're abusing the "last_update" field here?
     $feedsub->whereAdd(sprintf('last_update < "%s"', common_sql_date(time() - $args['interval'])));
     $feedsub->find();
     $qm = QueueManager::get();
     while ($feedsub->fetch()) {
         $orig = clone $feedsub;
         $item = array('id' => $feedsub->id);
         $qm->enqueue($item, self::QUEUE_CHECK);
         $feedsub->last_update = common_sql_now();
         $feedsub->update($orig);
     }
 }

 /**
  * Handler for POST content updates from the hub
  */
 function handlePost()
 {
     $feedid = $this->arg('feed');
     common_log(LOG_INFO, "POST for feed id {$feedid}");
     if (!$feedid) {
         throw new ServerException('Empty or invalid feed id', 400);
     }
     $feedsub = FeedSub::staticGet('id', $feedid);
     if (!$feedsub) {
         throw new ServerException('Unknown PuSH feed id ' . $feedid, 400);
     }
     $hmac = '';
     if (isset($_SERVER['HTTP_X_HUB_SIGNATURE'])) {
         $hmac = $_SERVER['HTTP_X_HUB_SIGNATURE'];
     }
     $post = file_get_contents('php://input');
     // Queue this to a background process; we should return
     // as quickly as possible from a distribution POST.
     // If queues are disabled this'll process immediately.
     $data = array('feedsub_id' => $feedsub->id, 'post' => $post, 'hmac' => $hmac);
     $qm = QueueManager::get();
     $qm->enqueue($data, 'pushin');
 }

 function handle($object)
 {
     list($user, $remote, $password) = $object;
     $remote = Discovery::normalize($remote);
     $oprofile = Ostatus_profile::ensureProfileURI($remote);
     if (empty($oprofile)) {
         // TRANS: Exception thrown when an account could not be located when it should be moved.
         // TRANS: %s is the remote site.
         throw new Exception(sprintf(_("Cannot locate account %s."), $remote));
     }
     list($svcDocUrl, $username) = self::getServiceDocument($remote);
     $sink = new ActivitySink($svcDocUrl, $username, $password);
     $this->log(LOG_INFO, "Moving user {$user->nickname} " . "to {$remote}.");
     $stream = new UserActivityStream($user);
     // Reverse activities to run in correct chron order
     $acts = array_reverse($stream->activities);
     $this->log(LOG_INFO, "Got " . count($acts) . " activities " . "for {$user->nickname}.");
     $qm = QueueManager::get();
     foreach ($acts as $act) {
         $qm->enqueue(array($act, $sink, $user->uri, $remote), 'actmove');
     }
     $this->log(LOG_INFO, "Finished moving user {$user->nickname} " . "to {$remote}.");
 }

 /**
  * Queue a Salmon notification for later. If queues are disabled we'll
  * send immediately but won't get the return value.
  *
  * @param mixed $entry XML string, Notice, or Activity
  * @return boolean success
  */
 public function notifyDeferred($entry, $actor)
 {
     if ($this->salmonuri) {
         common_debug("OSTATUS: user {$actor->getNickname()} ({$actor->getID()}) wants to ping {$this->localProfile()->getNickname()} on {$this->salmonuri}");
         $data = array('salmonuri' => $this->salmonuri, 'entry' => $this->notifyPrepXml($entry), 'actor' => $actor->getID(), 'target' => $this->localProfile()->getID());
         $qm = QueueManager::get();
         return $qm->enqueue($data, 'salmon');
     }
     return false;
 }

 /**
  * Actually delete a user.
  *
  * @return void
  */
 function handlePost()
 {
     if (Event::handle('StartDeleteUser', array($this, $this->user))) {
         // Mark the account as deleted and shove low-level deletion tasks
         // to background queues. Removing a lot of posts can take a while...
         if (!$this->user->hasRole(Profile_role::DELETED)) {
             $this->user->grantRole(Profile_role::DELETED);
         }
         $qm = QueueManager::get();
         $qm->enqueue($this->user, 'deluser');
         Event::handle('EndDeleteUser', array($this, $this->user));
     }
 }

        exit(1);
    }
    if (!file_exists($filename)) {
        // TRANS: Exception thrown when a file upload cannot be found.
        // TRANS: %s is the file that could not be found.
        throw new Exception(sprintf(_m('No such file "%s".'), $filename));
    }
    if (!is_file($filename)) {
        // TRANS: Exception thrown when a file upload is incorrect.
        // TRANS: %s is the irregular file.
        throw new Exception(sprintf(_m('Not a regular file: "%s".'), $filename));
    }
    if (!is_readable($filename)) {
        // TRANS: Exception thrown when a file upload is not readable.
        // TRANS: %s is the file that could not be read.
        throw new Exception(sprintf(_m('File "%s" not readable.'), $filename));
    }
    // TRANS: %s is the filename that contains a backup for a user.
    printfv(_m('Getting backup from file "%s".') . "\n", $filename);
    $html = file_get_contents($filename);
    return $html;
}
try {
    $user = getUser();
    $html = getBookmarksFile();
    $qm = QueueManager::get();
    $qm->enqueue(array($user, $html), 'dlcsback');
} catch (Exception $e) {
    print $e->getMessage() . "\n";
    exit(1);
}

 /**
  * Import a single bookmark
  *
  * Takes a <dt>/<dd> pair. The <dt> has a single
  * <a> in it with some non-standard attributes.
  *
  * A <dt><dt><dd> sequence will appear as a <dt> with
  * anothe <dt> as a child. We handle this case recursively.
  *
  * @param User       $user User to import data as
  * @param DOMElement $dt   <dt> element
  * @param DOMElement $dd   <dd> element
  *
  * @return Notice imported notice
  */
 function importBookmark($user, $dt, $dd = null)
 {
     $as = $dt->getElementsByTagName('a');
     if ($as->length == 0) {
         // TRANS: Client exception thrown when a bookmark in an import file is incorrectly formatted.
         throw new ClientException(_m("No <A> tag in a <DT>."));
     }
     $a = $as->item(0);
     $private = $a->getAttribute('private');
     if ($private != 0) {
         // TRANS: Client exception thrown when a bookmark in an import file is private.
         throw new ClientException(_m('Skipping private bookmark.'));
     }
     if (!empty($dd)) {
         $description = $dd->nodeValue;
     } else {
         $description = null;
     }
     $addDate = $a->getAttribute('add_date');
     $data = array('profile_id' => $user->id, 'title' => $a->nodeValue, 'description' => $description, 'url' => $a->getAttribute('href'), 'tags' => $a->getAttribute('tags'), 'created' => common_sql_date(intval($addDate)));
     $qm = QueueManager::get();
     $qm->enqueue($data, 'dlcsbkmk');
 }

 /**
  * Queue a file for restoration
  *
  * Uses the UserActivityStream class; may take a long time!
  *
  * @return void
  */
 function restoreAccount()
 {
     $this->checkSessionToken();
     if (!isset($_FILES['restorefile']['error'])) {
         // TRANS: Client exception displayed trying to restore an account while something went wrong uploading a file.
         throw new ClientException(_('No uploaded file.'));
     }
     switch ($_FILES['restorefile']['error']) {
         case UPLOAD_ERR_OK:
             // success, jump out
             break;
         case UPLOAD_ERR_INI_SIZE:
             // TRANS: Client exception thrown when an uploaded file is larger than set in php.ini.
             throw new ClientException(_('The uploaded file exceeds the ' . 'upload_max_filesize directive in php.ini.'));
             return;
         case UPLOAD_ERR_FORM_SIZE:
             throw new ClientException(_('The uploaded file exceeds the MAX_FILE_SIZE directive' . ' that was specified in the HTML form.'));
             return;
         case UPLOAD_ERR_PARTIAL:
             @unlink($_FILES['restorefile']['tmp_name']);
             // TRANS: Client exception.
             throw new ClientException(_('The uploaded file was only' . ' partially uploaded.'));
             return;
         case UPLOAD_ERR_NO_FILE:
             // TRANS: Client exception. No file; probably just a non-AJAX submission.
             throw new ClientException(_('No uploaded file.'));
             return;
         case UPLOAD_ERR_NO_TMP_DIR:
             // TRANS: Client exception thrown when a temporary folder is not present to store a file upload.
             throw new ClientException(_('Missing a temporary folder.'));
             return;
         case UPLOAD_ERR_CANT_WRITE:
             // TRANS: Client exception thrown when writing to disk is not possible during a file upload operation.
             throw new ClientException(_('Failed to write file to disk.'));
             return;
         case UPLOAD_ERR_EXTENSION:
             // TRANS: Client exception thrown when a file upload operation has been stopped by an extension.
             throw new ClientException(_('File upload stopped by extension.'));
             return;
         default:
             common_log(LOG_ERR, __METHOD__ . ": Unknown upload error " . $_FILES['restorefile']['error']);
             // TRANS: Client exception thrown when a file upload operation has failed with an unknown reason.
             throw new ClientException(_('System error uploading file.'));
             return;
     }
     $filename = $_FILES['restorefile']['tmp_name'];
     try {
         if (!file_exists($filename)) {
             // TRANS: Server exception thrown when an expected file upload could not be found.
             throw new ServerException(_("No such file '{$filename}'."));
         }
         if (!is_file($filename)) {
             // TRANS: Server exception thrown when an expected file upload is not an actual file.
             throw new ServerException(_("Not a regular file: '{$filename}'."));
         }
         if (!is_readable($filename)) {
             // TRANS: Server exception thrown when an expected file upload could not be read.
             throw new ServerException(_("File '{$filename}' not readable."));
         }
         common_debug(sprintf("Getting backup from file '%s'.", $filename));
         $xml = file_get_contents($filename);
         // This check is costly but we should probably give
         // the user some info ahead of time.
         $doc = new DOMDocument();
         // Disable PHP warnings so we don't spew low-level XML errors to output...
         // would be nice if we can just get exceptions instead.
         $old_err = error_reporting();
         error_reporting($old_err & ~E_WARNING);
         $doc->loadXML($xml);
         error_reporting($old_err);
         $feed = $doc->documentElement;
         if (!$feed || $feed->namespaceURI != Activity::ATOM || $feed->localName != 'feed') {
             // TRANS: Client exception thrown when a feed is not an Atom feed.
             throw new ClientException(_("Not an Atom feed."));
         }
         // Enqueue for processing.
         $qm = QueueManager::get();
         $qm->enqueue(array(common_current_user(), $xml, false), 'feedimp');
         if ($qm instanceof UnQueueManager) {
             // No active queuing means we've actually just completed the job!
             $this->success = true;
         } else {
             // We've fed data into background queues, and it's probably still running.
             $this->inprogress = true;
         }
         $this->showPage();
     } catch (Exception $e) {
         // Delete the file and re-throw
         @unlink($_FILES['restorefile']['tmp_name']);
         throw $e;
     }
 }