$envelope = array('site' => $siteNickname ? $siteNickname : common_config('site', 'nickname'),
'handler' => $queue,
'payload' => $this->encode($object));
- $msg = serialize($envelope);
+ $msg = base64_encode(serialize($envelope));
$props = array('created' => common_sql_now());
if ($this->isPersistent($queue)) {
protected function handleItem($frame)
{
$host = $this->cons[$this->defaultIdx]->getServer();
- $message = unserialize($frame->body);
+ $message = unserialize(base64_decode($frame->body));
if ($message === false) {
$this->_log(LOG_ERR, "Can't unserialize frame: {$frame->body}");
$site = $message['site'];
$queue = $message['handler'];
- if ($this->isDeadletter($frame, $message)) {
+ if ($this->isDeadLetter($frame, $message)) {
$this->stats('deadletter', $queue);
return false;
}
// @fixme detect failing site switches
$this->switchSite($site);
- $item = $this->decode($message['payload']);
- if (empty($item)) {
+ try {
+ $item = $this->decode($message['payload']);
+ } catch (Exception $e) {
$this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host");
$this->stats('baditem', $queue);
return false;
$frame->headers['created'] . " in queue $queue from $host";
$this->_log(LOG_DEBUG, "Dequeued $info");
- $handler = $this->getHandler($queue);
- if (!$handler) {
+ try {
+ $handler = $this->getHandler($queue);
+ $ok = $handler->handle($item);
+ } catch (NoQueueHandlerException $e) {
$this->_log(LOG_ERR, "Missing handler class; skipping $info");
$this->stats('badhandler', $queue);
return false;
- }
-
- try {
- $ok = $handler->handle($item);
} catch (Exception $e) {
$this->_log(LOG_ERR, "Exception on queue $queue: " . $e->getMessage());
$ok = false;
*/
function switchSite($site)
{
- if ($site != StatusNet::currentSite()) {
+ if ($site != GNUsocial::currentSite()) {
$this->stats('switch');
- StatusNet::switchSite($site);
+ GNUsocial::switchSite($site);
$this->initialize();
}
}
protected function queueName($queue)
{
$group = $this->queueGroup($queue);
- $site = StatusNet::currentSite();
+ $site = GNUsocial::currentSite();
$specs = array("$group/$queue/$site",
"$group/$queue");