123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177 |
- <?php
- class LiberalStomp extends Stomp
- {
-
- function getSocket()
- {
- return $this->_socket;
- }
-
- function getServer()
- {
- $idx = $this->_currentHost;
- if ($idx >= 0) {
- $host = $this->_hosts[$idx];
- return "$host[0]:$host[1]";
- } else {
- return '[unconnected]';
- }
- }
-
- protected function _makeConnection ()
- {
- parent::_makeConnection();
- stream_set_blocking($this->_socket, 0);
- }
-
- public function readFrames ()
- {
- if (!$this->hasFrameToRead()) {
- return false;
- }
-
- $rb = 1024;
- $data = '';
- $end = false;
- $frames = array();
- do {
-
-
- $read = fread($this->_socket, $rb);
- if ($read === false || ($read === '' && feof($this->_socket))) {
-
- throw new StompException("Error reading");
-
-
-
- }
- $data .= $read;
- if (strpos($data, "\x00") !== false) {
-
-
- $data = str_replace("\x00\n", "\x00", $data);
- $chunks = explode("\x00", $data);
- $data = array_pop($chunks);
- $frames = array_merge($frames, $chunks);
- if ($data == '') {
-
- break;
- } else {
-
- }
- }
-
-
- } while (true);
- return array_map(array($this, 'parseFrame'), $frames);
- }
-
-
- function parseFrame($data)
- {
- 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;
- }
-
- 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();
-
- stream_set_blocking($this->_socket, 1);
- $r = fwrite($this->_socket, $data, strlen($data));
- stream_set_blocking($this->_socket, 0);
- if ($r === false || $r == 0) {
- $this->_reconnect();
- $this->_writeFrame($stompFrame);
- }
- }
- }
|