diff options
Diffstat (limited to 'lib/stompqueuemanager.php')
-rw-r--r-- | lib/stompqueuemanager.php | 44 |
1 files changed, 13 insertions, 31 deletions
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php index 6730cd213..cd62c25bd 100644 --- a/lib/stompqueuemanager.php +++ b/lib/stompqueuemanager.php @@ -107,9 +107,10 @@ class StompQueueManager extends QueueManager $message .= ':' . $param; } $this->_connect(); - $result = $this->_send($this->control, - $message, - array ('created' => common_sql_now())); + $con = $this->cons[$this->defaultIdx]; + $result = $con->send($this->control, + $message, + array ('created' => common_sql_now())); if ($result) { $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message"); return true; @@ -368,17 +369,10 @@ class StompQueueManager extends QueueManager foreach ($this->cons as $i => $con) { if ($con) { $this->rollback($i); - $con->unsubscribe($this->control); + $con->disconnect(); + $this->cons[$i] = null; } } - if ($this->sites) { - foreach ($this->sites as $server) { - StatusNet::init($server); - $this->doUnsubscribe(); - } - } else { - $this->doUnsubscribe(); - } return true; } @@ -555,26 +549,14 @@ class StompQueueManager extends QueueManager } $host = $this->cons[$idx]->getServer(); - if (is_numeric($frame->body)) { - $id = intval($frame->body); - $info = "notice $id posted at {$frame->headers['created']} in queue $queue from $host"; - - $notice = Notice::staticGet('id', $id); - if (empty($notice)) { - $this->_log(LOG_WARNING, "Skipping missing $info"); - $this->ack($idx, $frame); - $this->commit($idx); - $this->begin($idx); - $this->stats('badnotice', $queue); - return false; - } - - $item = $notice; - } else { - // @fixme should we serialize, or json, or what here? - $info = "string posted at {$frame->headers['created']} in queue $queue from $host"; - $item = $frame->body; + $item = $this->decode($frame->body); + if (empty($item)) { + $this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host"); + return true; } + $info = $this->logrep($item) . " posted at " . + $frame->headers['created'] . " in queue $queue from $host"; + $this->_log(LOG_DEBUG, "Dequeued $info"); $handler = $this->getHandler($queue); if (!$handler) { |