123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614 |
- <?php
- require_once 'Stomp/Frame.php';
- class Stomp
- {
-
- public $sync = false;
-
- public $prefetchSize = 1;
-
-
- 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;
-
-
- public function __construct ($brokerUri)
- {
- $this->_brokerUri = $brokerUri;
- $this->_init();
- }
-
- 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}");
- }
- }
-
- 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");
- }
- }
-
- protected function _makeConnection ()
- {
- if (count($this->_hosts) == 0) {
- require_once 'Stomp/Exception.php';
- throw new StompException("No broker defined");
- }
-
-
- $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");
- }
- }
-
- 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");
- }
- }
- }
-
-
- public function isConnected ()
- {
- return !empty($this->_sessionId) && is_resource($this->_socket);
- }
-
- public function getSessionId()
- {
- return $this->_sessionId;
- }
-
- 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);
- }
-
- protected function _prepareReceipt (StompFrame $frame, $sync)
- {
- $receive = $this->sync;
- if ($sync !== null) {
- $receive = $sync;
- }
- if ($receive == true) {
- $frame->headers['receipt'] = md5(microtime());
- }
- }
-
- protected function _waitForReceipt (StompFrame $frame, $sync)
- {
- $receive = $this->sync;
- if ($sync !== null) {
- $receive = $sync;
- }
- if ($receive == true) {
- $id = (isset($frame->headers['receipt'])) ? $frame->headers['receipt'] : null;
- if ($id == null) {
- return true;
- }
- $frame = $this->readFrame();
- if ($frame instanceof StompFrame && $frame->command == 'RECEIPT') {
- if ($frame->headers['receipt-id'] == $id) {
- return true;
- } else {
- require_once 'Stomp/Exception.php';
- throw new StompException("Unexpected receipt id {$frame->headers['receipt-id']}", 0, $frame->body);
- }
- } else {
- require_once 'Stomp/Exception.php';
- if ($frame instanceof StompFrame) {
- throw new StompException("Unexpected command {$frame->command}", 0, $frame->body);
- } else {
- throw new StompException("Receipt not received");
- }
- }
- }
- return true;
- }
-
- public function subscribe ($destination, $properties = null, $sync = null)
- {
- $headers = array('ack' => 'client');
- $headers['activemq.prefetchSize'] = $this->prefetchSize;
- if ($this->clientId != null) {
- $headers["activemq.subcriptionName"] = $this->clientId;
- }
- if (isset($properties)) {
- foreach ($properties as $name => $value) {
- $headers[$name] = $value;
- }
- }
- $headers['destination'] = $destination;
- $frame = new StompFrame('SUBSCRIBE', $headers);
- $this->_prepareReceipt($frame, $sync);
- $this->_writeFrame($frame);
- if ($this->_waitForReceipt($frame, $sync) == true) {
- $this->_subscriptions[$destination] = $properties;
- return true;
- } else {
- return false;
- }
- }
-
- public function unsubscribe ($destination, $properties = null, $sync = null)
- {
- $headers = array();
- if (isset($properties)) {
- foreach ($properties as $name => $value) {
- $headers[$name] = $value;
- }
- }
- $headers['destination'] = $destination;
- $frame = new StompFrame('UNSUBSCRIBE', $headers);
- $this->_prepareReceipt($frame, $sync);
- $this->_writeFrame($frame);
- if ($this->_waitForReceipt($frame, $sync) == true) {
- unset($this->_subscriptions[$destination]);
- return true;
- } else {
- return false;
- }
- }
-
- public function begin ($transactionId = null, $sync = null)
- {
- $headers = array();
- if (isset($transactionId)) {
- $headers['transaction'] = $transactionId;
- }
- $frame = new StompFrame('BEGIN', $headers);
- $this->_prepareReceipt($frame, $sync);
- $this->_writeFrame($frame);
- return $this->_waitForReceipt($frame, $sync);
- }
-
- public function commit ($transactionId = null, $sync = null)
- {
- $headers = array();
- if (isset($transactionId)) {
- $headers['transaction'] = $transactionId;
- }
- $frame = new StompFrame('COMMIT', $headers);
- $this->_prepareReceipt($frame, $sync);
- $this->_writeFrame($frame);
- return $this->_waitForReceipt($frame, $sync);
- }
-
- public function abort ($transactionId = null, $sync = null)
- {
- $headers = array();
- if (isset($transactionId)) {
- $headers['transaction'] = $transactionId;
- }
- $frame = new StompFrame('ABORT', $headers);
- $this->_prepareReceipt($frame, $sync);
- $this->_writeFrame($frame);
- return $this->_waitForReceipt($frame, $sync);
- }
-
- public function ack ($message, $transactionId = null)
- {
- if ($message instanceof StompFrame) {
- $headers = $message->headers;
- if (isset($transactionId)) {
- $headers['transaction'] = $transactionId;
- }
- $frame = new StompFrame('ACK', $headers);
- $this->_writeFrame($frame);
- return true;
- } else {
- $headers = array();
- if (isset($transactionId)) {
- $headers['transaction'] = $transactionId;
- }
- $headers['message-id'] = $message;
- $frame = new StompFrame('ACK', $headers);
- $this->_writeFrame($frame);
- return true;
- }
- }
-
- public function disconnect ()
- {
- $headers = array();
- if ($this->clientId != null) {
- $headers["client-id"] = $this->clientId;
- }
- if (is_resource($this->_socket)) {
- $this->_writeFrame(new StompFrame('DISCONNECT', $headers));
- fclose($this->_socket);
- }
- $this->_socket = null;
- $this->_sessionId = null;
- $this->_currentHost = -1;
- $this->_subscriptions = array();
- $this->_username = '';
- $this->_password = '';
- }
-
- protected function _writeFrame (StompFrame $stompFrame)
- {
- if (!is_resource($this->_socket)) {
- require_once 'Stomp/Exception.php';
- throw new StompException('Socket connection hasn\'t been established');
- }
- $data = $stompFrame->__toString();
- $r = fwrite($this->_socket, $data, strlen($data));
- if ($r === false || $r == 0) {
- $this->_reconnect();
- $this->_writeFrame($stompFrame);
- }
- }
-
-
- public function setReadTimeout($seconds, $milliseconds = 0)
- {
- $this->_read_timeout_seconds = $seconds;
- $this->_read_timeout_milliseconds = $milliseconds;
- }
-
-
- public function readFrame ()
- {
- if (!$this->hasFrameToRead()) {
- return false;
- }
-
- $rb = 1024;
- $data = '';
- $end = false;
-
- do {
- $read = fread($this->_socket, $rb);
- if ($read === false) {
- $this->_reconnect();
- return $this->readFrame();
- }
- $data .= $read;
- if (strpos($data, "\x00") !== false) {
- $end = true;
- $data = rtrim($data, "\n");
- }
- $len = strlen($data);
- } while ($len < 2 || $end == false);
-
- list ($header, $body) = explode("\n\n", $data, 2);
- $header = explode("\n", $header);
- $headers = array();
- $command = null;
- foreach ($header as $v) {
- if (isset($command)) {
- list ($name, $value) = explode(':', $v, 2);
- $headers[$name] = $value;
- } else {
- $command = $v;
- }
- }
- $frame = new StompFrame($command, $headers, trim($body));
- if (isset($frame->headers['transformation']) && $frame->headers['transformation'] == 'jms-map-json') {
- require_once 'Stomp/Message/Map.php';
- return new StompMessageMap($frame);
- } else {
- return $frame;
- }
- return $frame;
- }
-
-
- public function hasFrameToRead()
- {
- $read = array($this->_socket);
- $write = null;
- $except = null;
-
- $has_frame_to_read = @stream_select($read, $write, $except, $this->_read_timeout_seconds, $this->_read_timeout_milliseconds);
-
- if ($has_frame_to_read !== false)
- $has_frame_to_read = count($read);
- if ($has_frame_to_read === false) {
- throw new StompException('Check failed to determine if the socket is readable');
- } else if ($has_frame_to_read > 0) {
- return true;
- } else {
- return false;
- }
- }
-
-
- protected function _reconnect ()
- {
- $subscriptions = $this->_subscriptions;
-
- $this->connect($this->_username, $this->_password);
- foreach ($subscriptions as $dest => $properties) {
- $this->subscribe($dest, $properties);
- }
- }
-
- public function __destruct()
- {
- $this->disconnect();
- }
- }
- ?>
|