ActiveMQ的PHP扩展stomp.php

<?php /** * * Copyright 2005-2006 The Apache Software Foundation * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ /* vim: set expandtab tabstop=3 shiftwidth=3: */ require_once 'Stomp/Frame.php'; /** * A Stomp Connection * * * @package Stomp * @author Hiram Chirino <hiram@hiramchirino.com> * @author Dejan Bosanac <dejan@nighttale.net> * @author Michael Caplan <mcaplan@labnet.net> * @version $Revision: 43 $ */ class Stomp { /** * Perform request synchronously * * @var boolean */ public $sync = false; /** * Default prefetch size * * @var int */ public $prefetchSize = 1; /** * Client id used for durable subscriptions * * @var string */ public $clientId = null; protected $_brokerUri = null; protected $_socket = null; protected $_hosts = array(); protected $_params = array(); protected $_subscriptions = array(); protected $_defaultPort = 61613; protected $_currentHost = - 1; protected $_attempts = 10; protected $_username = ''; protected $_password = ''; protected $_sessionId; protected $_read_timeout_seconds = 60; protected $_read_timeout_milliseconds = 0; protected $_connect_timeout_seconds = 60; /** * Constructor * * @param string $brokerUri Broker URL * @throws StompException */ public function __construct ($brokerUri) { $this->_brokerUri = $brokerUri; $this->_init(); } /** * Initialize connection * * @throws StompException */ protected function _init () { $pattern = "|^(([a-zA-Z]+)://)+\(*([a-zA-Z0-9\.:/i,-]+)\)*\??([a-zA-Z0-9=]*)$|i"; if (preg_match($pattern, $this->_brokerUri, $regs)) { $scheme = $regs[2]; $hosts = $regs[3]; $params = $regs[4]; if ($scheme != "failover") { $this->_processUrl($this->_brokerUri); } else { $urls = explode(",", $hosts); foreach ($urls as $url) { $this->_processUrl($url); } } if ($params != null) { parse_str($params, $this->_params); } } else { require_once 'Stomp/Exception.php'; throw new StompException("Bad Broker URL {$this->_brokerUri}"); } } /** * Process broker URL * * @param string $url Broker URL * @throws StompException * @return boolean */ protected function _processUrl ($url) { $parsed = parse_url($url); if ($parsed) { array_push($this->_hosts, array($parsed['host'] , $parsed['port'] , $parsed['scheme'])); } else { require_once 'Stomp/Exception.php'; throw new StompException("Bad Broker URL $url"); } } /** * Make socket connection to the server * * @throws StompException */ protected function _makeConnection () { if (count($this->_hosts) == 0) { require_once 'Stomp/Exception.php'; throw new StompException("No broker defined"); } // force disconnect, if previous established connection exists $this->disconnect(); $i = $this->_currentHost; $att = 0; $connected = false; $connect_errno = null; $connect_errstr = null; while (! $connected && $att ++ < $this->_attempts) { if (isset($this->_params['randomize']) && $this->_params['randomize'] == 'true') { $i = rand(0, count($this->_hosts) - 1); } else { $i = ($i + 1) % count($this->_hosts); } $broker = $this->_hosts[$i]; $host = $broker[0]; $port = $broker[1]; $scheme = $broker[2]; if ($port == null) { $port = $this->_defaultPort; } if ($this->_socket != null) { fclose($this->_socket); $this->_socket = null; } $this->_socket = @fsockopen($scheme . '://' . $host, $port, $connect_errno, $connect_errstr, $this->_connect_timeout_seconds); if (!is_resource($this->_socket) && $att >= $this->_attempts && !array_key_exists($i + 1, $this->_hosts)) { require_once 'Stomp/Exception.php'; throw new StompException("Could not connect to $host:$port ($att/{$this->_attempts})"); } else if (is_resource($this->_socket)) { $connected = true; $this->_currentHost = $i; break; } } if (! $connected) { require_once 'Stomp/Exception.php'; throw new StompException("Could not connect to a broker"); } } /** * Connect to server * * @param string $username * @param string $password * @return boolean * @throws StompException */ public function connect ($username = '', $password = '') { $this->_makeConnection(); if ($username != '') { $this->_username = $username; } if ($password != '') { $this->_password = $password; } $headers = array('login' => $this->_username , 'passcode' => $this->_password); if ($this->clientId != null) { $headers["client-id"] = $this->clientId; } $frame = new StompFrame("CONNECT", $headers); $this->_writeFrame($frame); $frame = $this->readFrame(); if ($frame instanceof StompFrame && $frame->command == 'CONNECTED') { $this->_sessionId = $frame->headers["session"]; return true; } else { require_once 'Stomp/Exception.php'; if ($frame instanceof StompFrame) { throw new StompException("Unexpected command: {$frame->command}", 0, $frame->body); } else { throw new StompException("Connection not acknowledged"); } } } /** * Check if client session has ben established * * @return boolean */ public function isConnected () { return !empty($this->_sessionId) && is_resource($this->_socket); } /** * Current stomp session ID * * @return string */ public function getSessionId() { return $this->_sessionId; } /** * Send a message to a destination in the messaging system * * @param string $destination Destination queue * @param string|StompFrame $msg Message * @param array $properties * @param boolean $sync Perform request synchronously * @return boolean */ public function send ($destination, $msg, $properties = array(), $sync = null) { if ($msg instanceof StompFrame) { $msg->headers['destination'] = $destination; if (is_array($properties)) $msg->headers = array_merge($msg->headers, $properties); $frame = $msg; } else { $headers = $properties; $headers['destination'] = $destination; $frame = new StompFrame('SEND', $headers, $msg); } $this->_prepareReceipt($frame, $sync); $this->_writeFrame($frame); return $this->_waitForReceipt($frame, $sync); } /** * Prepair frame receipt * * @param StompFrame $f