+ protected function _reconnect($idx)
+ {
+ try {
+ $con = $this->_doConnect($this->servers[$idx]);
+ } catch (Exception $e) {
+ $this->_log(LOG_ERR, $e->getMessage());
+ $con = null;
+ }
+ if ($con) {
+ $this->cons[$idx] = $con;
+ $this->disconnect[$idx] = null;
+
+ // now we have to listen to everything...
+ // @fixme refactor this nicer. :P
+ $host = $con->getServer();
+ $this->_log(LOG_INFO, "Resubscribing to $this->control on $host");
+ $con->subscribe($this->control);
+ foreach ($this->subscriptions as $site => $queues) {
+ foreach ($queues as $queue) {
+ $this->_log(LOG_INFO, "Resubscribing to $queue on $host");
+ $con->subscribe($queue);
+ }
+ }
+ $this->begin($idx);
+ } else {
+ // Try again later...
+ $this->disconnect[$idx] = time();
+ }
+ }
+
+ protected function _doConnect($server)
+ {
+ $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
+ $con = new LiberalStomp($server);
+
+ if ($con->connect($this->username, $this->password)) {
+ $this->_log(LOG_INFO, "Connected.");
+ } else {
+ $this->_log(LOG_ERR, 'Failed to connect to queue server');
+ throw new ServerException('Failed to connect to queue server');
+ }
+
+ return $con;
+ }
+