summaryrefslogtreecommitdiff
path: root/lib/stompqueuemanager.php
diff options
context:
space:
mode:
Diffstat (limited to 'lib/stompqueuemanager.php')
-rw-r--r--lib/stompqueuemanager.php81
1 files changed, 59 insertions, 22 deletions
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index 4cefba113..d13af3fa5 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -30,6 +30,14 @@
require_once 'Stomp.php';
+class LiberalStomp extends Stomp
+{
+ function getSocket()
+ {
+ return $this->_socket;
+ }
+}
+
class StompQueueManager
{
var $server = null;
@@ -50,7 +58,7 @@ class StompQueueManager
{
if (empty($this->con)) {
$this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
- $this->con = new Stomp($this->server);
+ $this->con = new LiberalStomp($this->server);
if ($this->con->connect($this->username, $this->password)) {
$this->_log(LOG_INFO, "Connected.");
@@ -94,37 +102,66 @@ class StompQueueManager
while (true) {
- $frame = $this->con->readFrame();
+ // Wait for something on one of our sockets
- if (!empty($frame)) {
- $notice = Notice::staticGet('id', $frame->body);
+ $stompsock = $this->con->getSocket();
- if (empty($notice)) {
- $this->_log(LOG_WARNING, 'Got ID '. $frame->body .' for non-existent notice in queue '. $queue);
- $this->con->ack($frame);
- } else {
- if ($handler->handle_notice($notice)) {
- $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
- $this->con->ack($frame);
- } else {
- $this->_log(LOG_WARNING, 'Failed handling notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
- // FIXME we probably shouldn't have to do
- // this kind of queue management ourselves
- $this->con->ack($frame);
- $this->enqueue($notice, $queue);
+ $handsocks = $handler->getSockets();
+
+ $socks = array_merge(array($stompsock), $handsocks);
+
+ $read = $socks;
+ $write = array();
+ $except = array();
+
+ $ready = stream_select($read, $write, $except, $handler->timeout(), 0);
+
+ if ($ready === false) {
+ $this->_log(LOG_ERR, "Error selecting on sockets");
+ } else if ($ready > 0) {
+ if (in_array($stompsock, $read)) {
+ $this->_handleNotice($queue, $handler);
+ }
+ foreach ($handsocks as $sock) {
+ if (in_array($sock, $read)) {
+ $handler->idle(QUEUE_HANDLER_HIT_IDLE);
+ break;
}
- unset($notice);
}
-
- unset($frame);
}
-
- $handler->idle(0);
}
$this->con->unsubscribe($this->_queueName($queue));
}
+ function _handleNotice($queue, $handler)
+ {
+ $frame = $this->con->readFrame();
+
+ if (!empty($frame)) {
+ $notice = Notice::staticGet('id', $frame->body);
+
+ if (empty($notice)) {
+ $this->_log(LOG_WARNING, 'Got ID '. $frame->body .' for non-existent notice in queue '. $queue);
+ $this->con->ack($frame);
+ } else {
+ if ($handler->handle_notice($notice)) {
+ $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
+ $this->con->ack($frame);
+ } else {
+ $this->_log(LOG_WARNING, 'Failed handling notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
+ // FIXME we probably shouldn't have to do
+ // this kind of queue management ourselves
+ $this->con->ack($frame);
+ $this->enqueue($notice, $queue);
+ }
+ unset($notice);
+ }
+
+ unset($frame);
+ }
+ }
+
function _queueName($queue)
{
return common_config('queue', 'queue_basename') . $queue;