code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Broker\RabbitMq;

use Stomp\Exception\StompException;
use Stomp\Protocol\Protocol;
use Stomp\Transport\Frame;
use Stomp\Protocol\Version;

/**
 * RabbitMq Stomp dialect.
 *
 *
 * @package Stomp
 * @author Hiram Chirino <hiram@hiramchirino.com>
 * @author Dejan Bosanac <dejan@nighttale.net>
 * @author Michael Caplan <mcaplan@labnet.net>
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class RabbitMq extends Protocol
{

    /**
     * Prefetch Size for subscriptions.
     *
     * @var int
     */

    private $prefetchCount = 1;

    /**
     * RabbitMq subscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param string $ack
     * @param string $selector
     * @param boolean|false $durable durable subscription
     * @return Frame
     */
    public function getSubscribeFrame(
        $destination,
        $subscriptionId = null,
        $ack = 'auto',
        $selector = null,
        $durable = false
    ) {
        $frame = parent::getSubscribeFrame($destination, $subscriptionId, $ack, $selector);
        $frame['prefetch-count'] = $this->prefetchCount;
        if ($durable) {
            $frame['persistent'] = 'true';
        }
        return $frame;
    }

    /**
     * RabbitMq unsubscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param bool|false $durable
     * @return \Stomp\Transport\Frame
     */
    public function getUnsubscribeFrame($destination, $subscriptionId = null, $durable = false)
    {
        $frame = parent::getUnsubscribeFrame($destination, $subscriptionId);
        if ($durable) {
            $frame['persistent'] = 'true';
        }
        return $frame;
    }

    /**
     * Prefetch Count for subscriptions
     *
     * @return int
     */
    public function getPrefetchCount()
    {
        return $this->prefetchCount;
    }

    /**
     * Prefetch Count for subscriptions
     *
     * @param int $prefetchCount
     */
    public function setPrefetchCount($prefetchCount)
    {
        $this->prefetchCount = $prefetchCount;
    }

    /**
     * Get message not acknowledge frame.
     *
     * @param \Stomp\Transport\Frame $frame
     * @param string $transactionId
     * @param bool $requeue Requeue header supported on RabbitMQ >= 3.4, ignored in prior versions
     * @return \Stomp\Transport\Frame
     * @throws StompException
     */
    public function getNackFrame(Frame $frame, $transactionId = null, $requeue = null)
    {
        if ($this->getVersion() === Version::VERSION_1_0) {
            throw new StompException('Stomp Version 1.0 has no support for NACK Frames.');
        }
        $nack = $this->createFrame('NACK');
        if ($requeue !== null) {
            $nack->addHeaders(['requeue' => $requeue ? 'true' : 'false']);
        }
        $nack['transaction'] = $transactionId;
        if ($this->hasVersion(Version::VERSION_1_2)) {
            $nack['id'] = $frame->getMessageId();
        } else {
            $nack['message-id'] = $frame->getMessageId();
            if ($this->hasVersion(Version::VERSION_1_1)) {
                $nack['subscription'] = $frame['subscription'];
            }
        }

        $nack['message-id'] = $frame->getMessageId();
        return $nack;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp;

use Stomp\Exception\ConnectionException;
use Stomp\Exception\MissingReceiptException;
use Stomp\Exception\StompException;
use Stomp\Exception\UnexpectedResponseException;
use Stomp\Network\Connection;
use Stomp\Protocol\Protocol;
use Stomp\Protocol\Version;
use Stomp\Transport\Frame;

/**
 * Stomp Client
 *
 * This is the minimal implementation of a Stomp Client, it allows to send and receive Frames using the Stomp Protocol.
 *
 * @package Stomp
 * @author Hiram Chirino <hiram@hiramchirino.com>
 * @author Dejan Bosanac <dejan@nighttale.net>
 * @author Michael Caplan <mcaplan@labnet.net>
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Client
{
    /**
     * Perform request synchronously
     *
     * @var boolean
     */
    private $sync = true;

    /**
     * Client id used for durable subscriptions
     *
     * @var string
     */
    private $clientId;

    /**
     * Connection session id
     *
     * @var string|null
     */
    private $sessionId;

    /**
     * Frames that have been read but not processed yet.
     *
     * @var Frame[]
     */
    private $unprocessedFrames = [];

    /**
     * @var Connection|null
     */
    private $connection;

    /**
     *
     * @var Protocol|null
     */
    private $protocol;

    /**
     * Seconds to wait for a receipt.
     *
     * @var float
     */
    private $receiptWait = 2;

    /**
     *
     * @var string
     */
    private $login;

    /**
     *
     * @var string
     */
    private $passcode;

    /**
     *
     * @var array
     */
    private $versions = [Version::VERSION_1_0, Version::VERSION_1_1, Version::VERSION_1_2];

    /**
     *
     * @var string
     */
    private $host;

    /**
     *
     * @var int[]
     */
    private $heartbeat = [0, 0];

    /**
     * @var bool
     */
    private $isConnecting = false;

    /**
     * Constructor
     *
     * @param string|Connection $broker Broker URL or a connection
     * @see Connection::__construct()
     */
    public function __construct($broker)
    {
        $this->connection = $broker instanceof Connection ? $broker : new Connection($broker);
    }

    /**
     * Configure versions to support.
     *
     * @param array $versions defaults to all client supported versions
     */
    public function setVersions(array $versions)
    {
        $this->versions = $versions;
    }

    /**
     * Configure the login to use.
     *
     * @param string $login
     * @param string $passcode
     */
    public function setLogin($login, $passcode)
    {
        $this->login = $login;
        $this->passcode = $passcode;
    }

    /**
     * Sets an fixed vhostname, which will be passed on connect as header['host'].
     *
     * (null = Default value is the hostname determined by connection.)
     *
     * @param string $host
     */
    public function setVhostname($host = null)
    {
        $this->host = $host;
    }

    /**
     * Set the desired heartbeat for the connection.
     *
     * A heartbeat is a specific message that will be send / received when no other data is send / received
     * within an interval - to indicate that the connection is still stable. If client and server agree on a beat and
     * the interval passes without any data activity / beats the connection will be considered as broken and closed.
     *
     * If you want to make sure that the server is still available, you should use the ServerAliveObserver
     * in combination with an requested server heartbeat interval.
     *
     * If you define a heartbeat for client side, you must assure that
     * your application will send data within the interval.
     * You can add \Stomp\Network\Observer\HeartbeatEmitter to your connection in order to send beats automatically.
     *
     * If you don't use HeartbeatEmitter you must either send messages within the interval
     * or make calls to Connection::sendAlive()
     *
     * @param int $send
     *   Number of milliseconds between expected sending of heartbeats. 0 means
     *   no heartbeats sent.
     * @param int $receive
     *   Number of milliseconds between expected receipt of heartbeats. 0 means
     *   no heartbeats expected. (not yet supported by this client)
     * @see \Stomp\Network\Observer\ServerAliveObserver
     * @see \Stomp\Network\Observer\HeartbeatEmitter
     * @see \Stomp\Network\Connection::sendAlive()
     */
    public function setHeartbeat($send = 0, $receive = 0)
    {
        $this->heartbeat = [$send, $receive];
    }

    /**
     * Connect to server
     *
     * @return boolean
     * @throws StompException
     * @see setVhostname
     */
    public function connect()
    {
        if ($this->isConnected()) {
            return true;
        }
        $this->isConnecting = true;
        $this->connection->connect();
        $this->connection->getParser()->legacyMode(true);
        $this->protocol = new Protocol($this->clientId);

        $this->host = $this->host ?: $this->connection->getHost();

        $connectFrame = $this->protocol->getConnectFrame(
            $this->login,
            $this->passcode,
            $this->versions,
            $this->host,
            $this->heartbeat
        );
        $this->sendFrame($connectFrame, false);

        if ($frame = $this->getConnectedFrame()) {
            $version = new Version($frame);

            if ($version->hasVersion(Version::VERSION_1_1)) {
                $this->connection->getParser()->legacyMode(false);
            }

            $this->sessionId = $frame['session'];
            $this->protocol = $version->getProtocol($this->clientId);
            $this->isConnecting = false;
            return true;
        }
        throw new ConnectionException('Connection not acknowledged');
    }

    /**
     * Returns the next available frame from the connection, respecting the connect timeout.
     *
     * @return null|Frame
     * @throws ConnectionException
     * @throws Exception\ErrorFrameException
     */
    private function getConnectedFrame()
    {
        $deadline = microtime(true) + $this->getConnection()->getConnectTimeout();
        do {
            if ($frame = $this->connection->readFrame()) {
                return $frame;
            }
        } while (microtime(true) <= $deadline);

        return null;
    }

    /**
     * Send a message to a destination in the messaging system
     *
     * @param string $destination Destination queue
     * @param string|Frame $msg Message
     * @param array $header
     * @param boolean $sync Perform request synchronously
     * @return boolean
     */
    public function send($destination, $msg, array $header = [], $sync = null)
    {
        if (!$msg instanceof Frame) {
            return $this->send($destination, new Frame('SEND', $header, $msg), [], $sync);
        }

        $msg->addHeaders($header);
        $msg['destination'] = $destination;
        return $this->sendFrame($msg, $sync);
    }

    /**
     * Send a frame.
     *
     * @param Frame $frame
     * @param boolean $sync
     * @return boolean
     */
    public function sendFrame(Frame $frame, $sync = null)
    {
        if (!$this->isConnecting && !$this->isConnected()) {
            $this->connect();
        }
        // determine if client was configured to write sync or not
        $writeSync = $sync !== null ? $sync : $this->sync;
        if ($writeSync) {
            return $this->sendFrameExpectingReceipt($frame);
        } else {
            return $this->connection->writeFrame($frame);
        }
    }

    /**
     * Write frame to server and expect an matching receipt frame
     *
     * @param Frame $stompFrame
     * @return bool
     */
    protected function sendFrameExpectingReceipt(Frame $stompFrame)
    {
        $receipt = md5(microtime());
        $stompFrame['receipt'] = $receipt;
        $this->connection->writeFrame($stompFrame);
        return $this->waitForReceipt($receipt);
    }

    /**
     * Wait for an receipt
     *
     * @param string $receipt
     * @return boolean
     * @throws UnexpectedResponseException If response has an invalid receipt.
     * @throws MissingReceiptException     If no receipt is received.
     */
    protected function waitForReceipt($receipt)
    {
        $stopAfter = $this->calculateReceiptWaitEnd();
        while (true) {
            if ($frame = $this->connection->readFrame()) {
                if ($frame->getCommand() == 'RECEIPT') {
                    if ($frame['receipt-id'] == $receipt) {
                        return true;
                    } else {
                        throw new UnexpectedResponseException($frame, sprintf('Expected receipt id %s', $receipt));
                    }
                } else {
                    $this->unprocessedFrames[] = $frame;
                }
            }
            if (microtime(true) >= $stopAfter) {
                break;
            }
        }
        throw new MissingReceiptException($receipt);
    }

    /**
     * Returns the timestamp with micro time to stop wait for a receipt.
     *
     * @return float
     */
    protected function calculateReceiptWaitEnd()
    {
        return microtime(true) + $this->receiptWait;
    }

    /**
     * Read response frame from server
     *
     * @return Frame|false when no frame to read
     */
    public function readFrame()
    {
        return array_shift($this->unprocessedFrames) ?: $this->connection->readFrame();
    }

    /**
     * Graceful disconnect from the server
     * @param bool $sync
     * @return void
     */
    public function disconnect($sync = false)
    {
        try {
            if ($this->connection && $this->connection->isConnected()) {
                if ($this->protocol) {
                    $this->sendFrame($this->protocol->getDisconnectFrame(), $sync);
                }
            }
        } catch (StompException $ex) {
            // nothing!
        }
        if ($this->connection) {
            $this->connection->disconnect();
        }

        $this->sessionId = null;
        $this->unprocessedFrames = [];
        $this->protocol = null;
        $this->isConnecting = false;
    }

    /**
     * Current stomp session ID
     *
     * @return string|null
     */
    public function getSessionId()
    {
        return $this->sessionId;
    }

    /**
     * Graceful object destruction
     *
     */
    public function __destruct()
    {
        $this->disconnect();
    }

    /**
     * Check if client session has ben established
     *
     * @return boolean
     */
    public function isConnected()
    {
        return !empty($this->sessionId) && $this->connection->isConnected();
    }

    /**
     * Get the used connection.
     *
     * @return Connection
     */
    public function getConnection()
    {
        return $this->connection;
    }

    /**
     * Get the currently used protocol.
     *
     * @return null|\Stomp\Protocol\Protocol
     */
    public function getProtocol()
    {
        if (!$this->isConnecting && !$this->isConnected()) {
            $this->connect();
        }
        return $this->protocol;
    }

    /**
     * @return string
     */
    public function getClientId()
    {
        return $this->clientId;
    }

    /**
     * @param string $clientId
     * @return Client
     */
    public function setClientId($clientId)
    {
        $this->clientId = $clientId;
        return $this;
    }

    /**
     * Set seconds to wait for a receipt.
     *
     * @param float $seconds
     */
    public function setReceiptWait($seconds)
    {
        $this->receiptWait = $seconds;
    }

    /**
     * Check if client runs in synchronized mode, which is the default operation mode.
     *
     * @return boolean
     */
    public function isSync()
    {
        return $this->sync;
    }

    /**
     * Toggle synchronized mode.
     *
     * @param boolean $sync
     */
    public function setSync($sync)
    {
        $this->sync = $sync;
    }
}

repo_name: ksuudilla/gnu

license: mit


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Exception;

/**
 * Any kind of connection problems.
 *
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ConnectionException extends StompException
{
    /**
     *
     * @var array
     */
    private $connectionInfo;

    /**
     *
     * @param string $info
     * @param array $connection
     * @param ConnectionException $previous
     */
    public function __construct($info, array $connection = [], ConnectionException $previous = null)
    {
        $this->connectionInfo = $connection;

        $host = ($previous ? $previous->getHostname() : null) ?: $this->getHostname();

        if ($host) {
            $info = sprintf('%s (Host: %s)', $info, $host);
        }
        parent::__construct($info, 0, $previous);
    }

    /**
     * Active used connection.
     *
     * @return array
     */
    public function getConnectionInfo()
    {
        return $this->connectionInfo;
    }

    protected function getHostname()
    {
        return isset($this->connectionInfo['host']) ? $this->connectionInfo['host'] : null;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Exception;

use Stomp\Transport\Frame;

/**
 * Stomp server send us an error frame.
 *
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ErrorFrameException extends StompException
{
    /**
     *
     * @var Frame
     */
    private $frame;

    /**
     *
     * @param Frame $frame
     */
    public function __construct(Frame $frame)
    {
        $this->frame = $frame;
        parent::__construct(
            sprintf('Error "%s"', $frame['message'])
        );
    }

    /**
     *
     * @return Frame
     */
    public function getFrame()
    {
        return $this->frame;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Exception;

/**
 * Exception that occurs, when a frame receipt was not received.
 *
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class MissingReceiptException extends StompException
{
    /**
     * @var string
     */
    private $receiptId;

    /**
     *
     * @param string $receiptId
     */
    public function __construct($receiptId)
    {
        $this->receiptId = $receiptId;
        parent::__construct(
            sprintf(
                'Missing receipt Frame for id "%s". Maybe the queue server is under heavy load. ' .
                'Try to increase timeouts.',
                $receiptId
            )
        );
    }

    /**
     * Expected receipt id.
     *
     * @return String
     */
    public function getReceiptId()
    {
        return $this->receiptId;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Exception;

use Exception;

/**
 * Base exception for all special stomp exceptions.
 *
 * @package Stomp
 */
class StompException extends Exception
{
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Exception;

use Stomp\Transport\Frame;

/**
 * Exception that occurs, when a frame / response was received that was not expected at this moment.
 *
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class UnexpectedResponseException extends StompException
{
    /**
     *
     * @var Frame
     */
    private $frame;

    /**
     *
     * @param Frame $frame
     * @param string $expectedInfo
     */
    public function __construct(Frame $frame, $expectedInfo)
    {
        $this->frame = $frame;
        parent::__construct(sprintf('Unexpected response received. %s', $expectedInfo));
    }

    /**
     *
     * @return Frame
     */
    public function getFrame()
    {
        return $this->frame;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network;

use Stomp\Exception\ConnectionException;
use Stomp\Exception\ErrorFrameException;
use Stomp\Network\Observer\ConnectionObserverCollection;
use Stomp\Transport\Frame;
use Stomp\Transport\Parser;

/**
 * A Stomp Connection
 *
 * @package Stomp
 * @author Hiram Chirino <hiram@hiramchirino.com>
 * @author Dejan Bosanac <dejan@nighttale.net>
 * @author Michael Caplan <mcaplan@labnet.net>
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Connection
{
    /**
     * Default ActiveMq port
     */
    const DEFAULT_PORT = 61613;

    /**
     * Host schemes.
     *
     * @var array[]
     */
    private $hosts = [];

    /**
     * Connection timeout in seconds.
     *
     * @var integer
     */
    private $connectTimeout;

    /**
     * Timeout (seconds) that are applied on write calls.
     *
     * @var integer
     */
    private $writeTimeout = 3;

    /**
     * Using persistent connection for creating socket
     *
     * @var bool
     */
    private $persistentConnection = false;

    /**
     * Connection read wait timeout.
     *
     * 0 => seconds
     * 1 => milliseconds
     *
     * @var array
     */
    private $readTimeout = [60, 0];

    /**
     * Connection options.
     *
     * @var array
     */
    private $params = [
        'randomize' => false // connect to one host from list in random order
    ];

    /**
     * Active connection resource.
     *
     * @var resource|null
     */
    private $connection;

    /**
     * Connected host info.
     *
     * @var array
     */
    private $activeHost = [];

    /**
     * Stream Context used for client connection
     *
     * @see http://php.net/manual/de/function.stream-context-create.php
     *
     * @var array
     */
    private $context = [];

    /**
     * Frame parser
     *
     * @var Parser
     */
    private $parser;

    /**
     * Host connected to.
     *
     * @var String
     */
    private $host;

    /**
     * @var ConnectionObserverCollection
     */
    private $observers;

    /**
     * Maximum number of bytes to write to a resource.
     *
     * @var int
     */
    private $maxWriteBytes = 8192;

    /**
     * Maximum number of bytes to read from a resource.
     *
     * @var int
     */
    private $maxReadBytes = 8192;

    /**
     * Alive Signal
     */
    const ALIVE = "\n";

    /**
     * @var callable|null
     */
    private $waitCallback;

    /**
     * Initialize connection
     *
     * Example broker uri
     * - Use only one broker uri: tcp://localhost:61614
     * - use failover in given order: failover://(tcp://localhost:61614,ssl://localhost:61612)
     * - use brokers in random order: failover://(tcp://localhost:61614,ssl://localhost:61612)?randomize=true
     *
     * @param string $brokerUri
     * @param integer $connectionTimeout in seconds
     * @param array $context stream context
     * @throws ConnectionException
     */
    public function __construct($brokerUri, $connectionTimeout = 1, array $context = [])
    {
        $this->parser = new Parser();
        $this->observers = new ConnectionObserverCollection();
        $this->parser->setObserver($this->observers);
        $this->connectTimeout = $connectionTimeout;
        $this->context = $context;
        $pattern = "|^(([a-zA-Z0-9]+)://)+\(*([a-zA-Z0-9\.:/i,-_]+)\)*\??([a-zA-Z0-9=&]*)$|i";
        if (preg_match($pattern, $brokerUri, $matches)) {
            $scheme = $matches[2];
            $hosts = $matches[3];
            $options = $matches[4];

            if ($options) {
                parse_str($options, $connectionOptions);
                $this->params = $connectionOptions + $this->params;
            }

            if ($scheme != 'failover') {
                $this->parseUrl($brokerUri);
            } else {
                $urls = explode(',', $hosts);
                foreach ($urls as $url) {
                    $this->parseUrl($url);
                }
            }
        }

        if (empty($this->hosts)) {
            throw new ConnectionException("Bad Broker URL {$brokerUri}. Check used scheme!");
        }
    }

    /**
     * Sets a wait callback that will be invoked when the connection is waiting for new data.
     *
     * This is a good place to call `pcntl_signal_dispatch()` if you need to ensure that your process signals in an
     * interval that is lower as read timeout. You should also return `false` in your callback if you don't want the
     * connection to continue waiting for data.
     *
     * @param callable|null $waitCallback
     */
    public function setWaitCallback($waitCallback)
    {
        if ($waitCallback !== null) {
            if (!is_callable($waitCallback)) {
                throw new \InvalidArgumentException('$waitCallback must be callable.');
            }
        }
        $this->waitCallback = $waitCallback;
    }

    /**
     * Returns the connect timeout in seconds.
     *
     * @return int
     */
    public function getConnectTimeout()
    {
        return $this->connectTimeout;
    }

    /**
     * Returns the collection of observers of this connection.
     *
     * @return ConnectionObserverCollection
     */
    public function getObservers()
    {
        return $this->observers;
    }

    /**
     * Parse a broker URL
     *
     * @param string $url Broker URL
     * @return void
     * @throws ConnectionException
     */
    private function parseUrl($url)
    {
        $parsed = parse_url($url);
        if ($parsed === false) {
            throw new ConnectionException('Unable to parse url '. $url);
        }
        array_push($this->hosts, $parsed + ['port' => '61613', 'scheme' => 'tcp']);
    }

    /**
     * Set the read timeout
     *
     * @param integer $seconds      seconds
     * @param integer $microseconds microseconds (1μs = 0.000001s, ex. 500ms = 500000)
     * @return void
     */
    public function setReadTimeout($seconds, $microseconds = 0)
    {
        $this->readTimeout[0] = $seconds;
        $this->readTimeout[1] = $microseconds;
    }

    /**
     * Returns the read timeout
     *
     * First element contains full seconds, second the microseconds part.
     *
     * @return array
     */
    public function getReadTimeout()
    {
        return $this->readTimeout;
    }

    /**
     * Set the write timeout
     *
     * @param int $writeTimeout seconds
     */
    public function setWriteTimeout($writeTimeout)
    {
        $this->writeTimeout = $writeTimeout;
    }

    /**
     * Set socket context
     *
     * @param array $context
     * @return void
     */
    public function setContext(array $context)
    {
        $this->context = $context;
    }

    /**
     * Set the maximum number of bytes to write to a resource
     *
     * This will be useful if you are suffering problems with OpenSSL or Amazon MQ
     *
     * @param int $maxWriteBytes bytes
     */
    public function setMaxWriteBytes($maxWriteBytes)
    {
        $this->maxWriteBytes = $maxWriteBytes;
    }

    /**
     * Set the maximum number of bytes to read from a resource
     *
     * This will be useful if you are suffering problems with OpenSSL or Amazon MQ
     *
     * @param int $maxReadBytes bytes
     */
    public function setMaxReadBytes($maxReadBytes)
    {
        $this->maxReadBytes = $maxReadBytes;
    }

    /**
     * Connect to an broker.
     *
     * @return boolean
     * @throws ConnectionException
     */
    public function connect()
    {
        if (!$this->isConnected()) {
            $this->connection = $this->getConnection();
        }
        return true;
    }

    /**
     * @param boolean $persistentConnection
     */
    public function setPersistentConnection($persistentConnection)
    {
        $this->persistentConnection = $persistentConnection;
    }

    /**
     * Get a connection.
     *
     * @return resource (stream)
     * @throws ConnectionException
     */
    protected function getConnection()
    {
        $hosts = $this->getHostList();

        while ($host = array_shift($hosts)) {
            try {
                return $this->connectSocket($host);
            } catch (ConnectionException $connectionException) {
                if (empty($hosts)) {
                    throw new ConnectionException("Could not connect to a broker", [], $connectionException);
                }
            }
        }
    }

    /**
     * Get the host list.
     *
     * @return array
     */
    protected function getHostList()
    {
        $hosts = array_values($this->hosts);
        if ($this->params['randomize']) {
            shuffle($hosts);
        }
        return $hosts;
    }

    /**
     * Try to connect to given host.
     *
     * @param array $host
     * @return resource (stream)
     * @throws ConnectionException if connection setup fails
     */
    protected function connectSocket(array $host)
    {
        $this->activeHost = $host;
        $errNo = null;
        $errStr = null;

        $context = stream_context_create($this->context);
        $flags = STREAM_CLIENT_CONNECT;
        if ($this->persistentConnection) {
            $flags |= STREAM_CLIENT_PERSISTENT;
        }
        $socket = @stream_socket_client(
            $host['scheme'] . '://' . $host['host'] . ':' . $host['port'],
            $errNo,
            $errStr,
            $this->connectTimeout,
            $flags,
            $context
        );

        if (!is_resource($socket)) {
            throw new ConnectionException(sprintf('Failed to connect. (%s: %s)', $errNo, $errStr), $host);
        }

        if (!@stream_set_blocking($socket, false)) {
            throw new ConnectionException('Failed to set non blocking mode for stream.', $host);
        }
        $this->host = $host['host'];
        return $socket;
    }

    /**
     * Connection established.
     *
     * @return boolean
     */
    public function isConnected()
    {
        return ($this->connection && is_resource($this->connection));
    }

    /**
     * Close connection.
     *
     * @return void
     */
    public function disconnect()
    {
        if ($this->isConnected()) {
            @stream_socket_shutdown($this->connection, STREAM_SHUT_RDWR);
        }
        $this->connection = null;
        $this->activeHost = [];
    }

    /**
     * Write frame to server.
     *
     * @param Frame $stompFrame
     * @return boolean
     * @throws ConnectionException
     */
    public function writeFrame(Frame $stompFrame)
    {
        if (!$this->isConnected()) {
            throw new ConnectionException('Not connected to any server.', $this->activeHost);
        }
        $this->writeData($stompFrame, $this->writeTimeout);
        $this->observers->sentFrame($stompFrame);
        return true;
    }

    /**
     * Write passed data to the stream, respecting passed timeout.
     *
     * @param Frame|string $stompFrame
     * @param float $timeout in seconds, supporting fractions
     * @throws ConnectionException
     */
    private function writeData($stompFrame, $timeout)
    {
        $data = (string) $stompFrame;
        $offset = 0;
        $size = strlen($data);
        $lastByteTime = microtime(true);
        do {
            $written = @fwrite($this->connection, substr($data, $offset), $this->maxWriteBytes);

            if ($written === false) {
                throw new ConnectionException('Was not possible to write frame!', $this->activeHost);
            }

            if ($written > 0) {
                // offset tracking
                $offset += $written;
                $lastByteTime = microtime(true);
            } else {
                // timeout tracking
                if ((microtime(true) - $lastByteTime) > $timeout) {
                    throw new ConnectionException(
                        'Was not possible to write frame! Write operation timed out.',
                        $this->activeHost
                    );
                }
            }
            // keep some time to breath
            if ($written < $size) {
                time_nanosleep(0, 2500000); // 2.5ms / 0.0025s
            }
        } while ($offset < $size);
    }

    /**
     * Try to read a frame from the server.
     *
     * @return Frame|false when no frame to read
     * @throws ConnectionException
     * @throws ErrorFrameException
     */
    public function readFrame()
    {
        // first we try to check the parser for any leftover frames
        if ($frame = $this->parser->nextFrame()) {
            return $this->onFrame($frame);
        }

        if (!$this->hasDataToRead()) {
            return false;
        }

        do {
            $read = @fread($this->connection, $this->maxReadBytes);
            if ($read === false) {
                throw new ConnectionException(sprintf('Was not possible to read data from stream.'), $this->activeHost);
            }

            // this can be caused by different events on the stream, ex. new data or any kind of signal
            // it also happens when a ssl socket was closed on the other side... so we need to test
            if ($read === '') {
                $this->observers->emptyRead();
                // again we give some time here
                // as this path is most likely indicating that the socket is not working anymore
                time_nanosleep(0, 5000000); // 5ms / 0.005s
                return false;
            }

            $this->parser->addData($read);

            if ($frame = $this->parser->nextFrame()) {
                return $this->onFrame($frame);
            }
        } while ($this->isDataOnStream());

        return false;
    }

    /**
     * The connection onFrame handler.
     *
     * @param Frame $frame
     * @return Frame
     * @throws ErrorFrameException
     */
    private function onFrame(Frame $frame)
    {
        if ($frame->isErrorFrame()) {
            throw new ErrorFrameException($frame);
        }
        return $frame;
    }

    /**
     * Check if connection has new data which can be read.
     *
     * This might wait until readTimeout is reached.
     *
     * @return boolean
     * @throws ConnectionException
     * @see Connection::setReadTimeout()
     */
    public function hasDataToRead()
    {
        if (!$this->isConnected()) {
            throw new ConnectionException('Not connected to any server.', $this->activeHost);
        }

        $isDataInBuffer = $this->connectionHasDataToRead($this->readTimeout[0], $this->readTimeout[1]);
        if (!$isDataInBuffer) {
            $this->observers->emptyBuffer();
        }
        return $isDataInBuffer;
    }

    /**
     * See if the connection has data left.
     *
     * If both timeout-parameters are set to 0, it will return immediately.
     *
     * @param int $timeoutSec Second-timeout part
     * @param int $timeoutMicros Microsecond-timeout part
     * @return bool
     * @throws ConnectionException
     */
    private function connectionHasDataToRead($timeoutSec, $timeoutMicros)
    {
        $timeout = microtime(true) + $timeoutSec + ($timeoutMicros ? $timeoutMicros / 1000000 : 0);
        while (($hasData = $this->isDataOnStream()) === false) {
            if ($timeout < microtime(true)) {
                return false;
            }
            if ($this->waitCallback) {
                if (call_user_func($this->waitCallback) === false) {
                    return false;
                }
            }

            $slept = time_nanosleep(0, 2500000); // 2.5ms / 0.0025s
            if (\is_array($slept)) {
                return false;
            }
        }

        return $hasData === true;
    }

    /**
     * Checks if there is readable data on the stream.
     *
     * Will return true if data is available, false if no data is detected and null if the operation was interrupted.
     *
     * @return bool|null
     * @throws ConnectionException
     */
    private function isDataOnStream()
    {
        $read = [$this->connection];
        $write = null;
        $except = null;
        $hasStreamInfo = @stream_select($read, $write, $except, 0);

        if ($hasStreamInfo === false) {
            // can return `false` if used in combination with `pcntl_signal` and lead to false errors here
            $error = error_get_last();
            if ($error && isset($error['message']) && stripos($error['message'], 'interrupted system call') === false) {
                throw new ConnectionException(
                    'Check failed to determine if the socket is readable.',
                    $this->activeHost
                );
            }
            return null;
        }

        return !empty($read);
    }

    /**
     * Returns the parser which is used by the connection.
     *
     * @return Parser
     */
    public function getParser()
    {
        return $this->parser;
    }

    /**
     * Returns the host the connection was established to.
     *
     * @return String
     */
    public function getHost()
    {
        return $this->host;
    }

    /**
     * Writes an "alive" message on the connection to indicate that the client is alive.
     *
     * @param float $timeout in seconds supporting fractions (microseconds)
     *
     * @return void
     * @throws ConnectionException
     */
    public function sendAlive($timeout = 1.0)
    {
        if ($this->isConnected()) {
            $this->writeData(self::ALIVE, $timeout);
        }
    }

    /**
     * Immediately releases all allocated resources when the connection object gets destroyed.
     *
     * This is especially important for long running processes.
     */
    public function __destruct()
    {
        $this->disconnect();
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer;

use Stomp\Transport\Frame;

/**
 * AbstractBeats base for heart beat observer.
 *
 * @package Stomp\Network\Observer
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
abstract class AbstractBeats implements ConnectionObserver
{
    /**
     * Frame from client that request a connection.
     */
    const FRAME_CLIENT_CONNECT = 'CONNECT';

    /**
     * Frame from server when a connection is established.
     */
    const FRAME_SERVER_CONNECTED = 'CONNECTED';
    /**
     * The beat interval that the client wants to use.
     *
     * @var integer
     */
    protected $intervalClient;
    /**
     * The beat interval that the server wants to use.
     *
     * @var integer
     */
    protected $intervalServer;
    /**
     * The timestamp that is known as the last time a beat was detected/issued.
     *
     * @var float
     */
    private $lastbeat;

    /**
     * The interval (seconds with microseconds as fraction) that will be used to detect a delay.
     *
     * @var float
     */
    private $intervalUsed;

    /**
     * Whenever the emitter is configured to send beats.
     *
     * @var bool
     */
    private $enabled = false;

    /**
     * A frame with heartbeat details was detected.
     *
     * Child class should set client or server interval property.
     *
     * @see $intervalClient
     * @see $intervalServer
     *
     * @param Frame $frame
     * @param array $beats
     * @return void
     */
    abstract protected function onHeartbeatFrame(Frame $frame, array $beats);

    /**
     * Called whenever the server send data.
     *
     * @return void
     */
    abstract protected function onServerActivity();

    /**
     * Must return the interval (ms) that should be used to detect a delay.
     *
     * @param integer $maximum agreement from client and server in milliseconds
     * @return float
     */
    abstract protected function calculateInterval($maximum);

    /**
     * Called whenever a activity is detected that was issued by the client.
     *
     * @return void
     */
    abstract protected function onClientActivity();

    /**
     * Something on the connection state could have changed.
     *
     * @return void
     */
    abstract protected function onPotentialConnectionStateActivity();

    /**
     * Delay was detected.
     *
     * @return void
     */
    abstract protected function onDelay();

    /**
     * Checks if the emitter was enabled.
     *
     * @return bool
     */
    public function isEnabled()
    {
        return $this->enabled;
    }

    /**
     * Returns if the emitter is in a state that indicates a delay.
     *
     * @return bool
     */
    public function isDelayed()
    {
        if ($this->enabled && $this->lastbeat) {
            $now = microtime(true);
            return ($now - $this->lastbeat > $this->intervalUsed);
        }
        return false;
    }

    /**
     * Returns the calculated interval for beats in seconds (with micro fraction).
     *
     * @return null|float
     */
    public function getInterval()
    {
        return $this->intervalUsed;
    }

    /**
     * Check if there is a delay and issue follow up tasks if so.
     *
     * @return void
     */
    protected function checkDelayed()
    {
        if ($this->isDelayed()) {
            $this->onDelay();
        }
    }

    /**
     * Outgoing activity event.
     *
     * @return void
     */
    protected function rememberActivity()
    {
        $this->lastbeat = microtime(true);
    }

    /**
     * Returns the heartbeat header.
     *
     * @param Frame $frame
     * @return array
     */
    private function getHeartbeats(Frame $frame)
    {
        $beats = $frame['heart-beat'];
        if ($beats) {
            return explode(',', $beats, 2);
        }
        return [0, 0];
    }

    /**
     * Enables the delay detection when preconditions are fulfilled.
     *
     * @param Frame $frame
     * @return void
     */
    private function enable(Frame $frame)
    {
        $this->onHeartbeatFrame($frame, $this->getHeartbeats($frame));
        if ($this->intervalServer && $this->intervalClient) {
            $intervalAgreement = $this->calculateInterval(max($this->intervalClient, $this->intervalServer));
            $this->intervalUsed = $intervalAgreement / 1000; // milli to micro
            if ($intervalAgreement) {
                $this->enabled = true;
                $this->rememberActivity();
            }
        }
    }

    /**
     * @inheritdoc
     */
    public function receivedFrame(Frame $frame)
    {
        if ($this->enabled) {
            $this->onServerActivity();
            return;
        }

        if ($frame->getCommand() === self::FRAME_SERVER_CONNECTED) {
            $this->enable($frame);
        }
    }

    /**
     * @inheritdoc
     */
    public function sentFrame(Frame $frame)
    {
        if ($this->enabled) {
            $this->onClientActivity();
            return;
        }
        if ($frame->getCommand() === self::FRAME_CLIENT_CONNECT) {
            $this->enable($frame);
        }
    }

    /**
     * @inheritdoc
     */
    public function emptyLineReceived()
    {
        $this->onServerActivity();
    }

    /**
     * @inheritdoc
     */
    public function emptyRead()
    {
        $this->onPotentialConnectionStateActivity();
    }

    /**
     * @inheritdoc
     */
    public function emptyBuffer()
    {
        $this->onServerActivity();
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer;

use Stomp\Transport\Frame;

/**
 * Interface ConnectionObserver defines the observable events on a connection.
 *
 * @package Stomp\Network\Observer
 */
interface ConnectionObserver
{
    /**
     * Indicates that during a read call no frame was received, but an EOL line.
     *
     * @return void
     */
    public function emptyLineReceived();

    /**
     * Indicates that the connection has no pending data.
     *
     * @return void
     */
    public function emptyBuffer();

    /**
     * Indicates that the connection tried to read signaled data, but no data was returned.
     *
     * @return void
     */
    public function emptyRead();

    /**
     * Indicates that a frame has been received from the server.
     *
     * @param Frame $frame that has been received
     * @return void
     */
    public function receivedFrame(Frame $frame);

    /**
     * Indicates that a frame has been sent to the server.
     *
     * @param Frame $frame
     * @return void
     */
    public function sentFrame(Frame $frame);
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer;

use Stomp\Transport\Frame;

/**
 * ConnectionObserverCollection a collection of connection observers.
 *
 * @package Stomp\Network\Observer
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ConnectionObserverCollection implements ConnectionObserver
{
    /**
     * @var ConnectionObserver[]
     */
    private $observers = [];

    /**
     * Adds new observers to the collection.
     *
     * @param ConnectionObserver $observer
     * @return ConnectionObserverCollection this collection
     */
    public function addObserver(ConnectionObserver $observer)
    {
        if (!in_array($observer, $this->observers, true)) {
            $this->observers[] = $observer;
        }
        return $this;
    }

    /**
     * Removes the observers from the collection.
     *
     * @param ConnectionObserver $observer
     * @return ConnectionObserverCollection this collection
     */
    public function removeObserver(ConnectionObserver $observer)
    {
        $index = array_search($observer, $this->observers, true);
        if ($index !== false) {
            unset($this->observers[$index]);
        }
        return $this;
    }

    /**
     * Returns the observers inside this collection.
     *
     * @return ConnectionObserver[]
     */
    public function getObservers()
    {
        return array_values($this->observers);
    }

    /**
     * Indicates that during a read call no frame was received, but an EOL line.
     *
     * @return void
     */
    public function emptyLineReceived()
    {
        foreach ($this->observers as $item) {
            $item->emptyLineReceived();
        }
    }

    /**
     * Indicates that a frame has been received.
     *
     * @param Frame $frame that has been received
     * @return void
     */
    public function receivedFrame(Frame $frame)
    {
        foreach ($this->observers as $item) {
            $item->receivedFrame($frame);
        }
    }

    /**
     * Indicates that a frame has been transmitted.
     *
     * @param Frame $frame
     * @return void
     */
    public function sentFrame(Frame $frame)
    {
        foreach ($this->observers as $item) {
            $item->sentFrame($frame);
        }
    }

    /**
     * Indicates that the connection has no pending data.
     *
     * @return void
     */
    public function emptyBuffer()
    {
        foreach ($this->observers as $item) {
            $item->emptyBuffer();
        }
    }

    /**
     * Indicates that the connection tried to read signaled data, but no data was returned.
     *
     * @return void
     */
    public function emptyRead()
    {
        foreach ($this->observers as $item) {
            $item->emptyRead();
        }
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer\Exception;

/**
 * HeartbeatException indicate that heartbeats where not send or received as expected.
 *
 * @package src\Network\Observer\Exception
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class HeartbeatException extends \RuntimeException
{
    /**
     * ClientHeartbeatException constructor.
     *
     * @param string $message
     * @param \Exception|null $previous
     */
    public function __construct($message, \Exception $previous = null)
    {
        parent::__construct($message, 0, $previous);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer;

use Stomp\Exception\ConnectionException;
use Stomp\Network\Connection;
use Stomp\Network\Observer\Exception\HeartbeatException;
use Stomp\Transport\Frame;

/**
 * HeartbeatEmitter a very basic heartbeat emitter that sends beats from client side.
 *
 * Use this if you can guarantee that your client processes workloads within a given interval. This allows the server to
 * detect that your client is down when it fails sending heartbeats, your client will also fail with exception when the
 * server is not longer receiving heartbeats.
 *
 * If your client needs a unknown runtime to process Messages you should check ServerAliveObserver.
 *
 * @example $client->setHeartbeat(2000, 0); // indicate that we would send beats within a 2 second interval
 *          $emitter = new HeartbeatEmitter($client->getConnection());
 *          $client->getConnection()->getObservers()->addObserver($emitter);
 *
 * @see ServerAliveObserver
 * @package Stomp\Network\Observer\Heartbeat
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class HeartbeatEmitter extends AbstractBeats
{
    /**
     * @var Connection
     */
    private $connection;

    /**
     * Defines the percentage amount of the calculated interval that will be used without emitting a beat.
     *
     * @var float
     */
    private $intervalUsage;

    /**
     * Enables the pessimistic mode of the emitter, causing alive messages when we receive nothing from the socket.
     *
     * @var bool
     */
    private $pessimistic = false;

    /**
     * Emitter constructor.
     *
     * What is the interval usage?
     * The usage (percentage) defines the amount of agreed beat interval time,
     * that is allowed to pass before the emitter will send a beat.
     *
     * A higher value increases the risk that a beat is send after a timeout has occurred.
     * A lower value increases the beats and adds overhead to the connection.
     *
     * @param Connection $connection
     * @param float $intervalUsage
     */
    public function __construct(Connection $connection, $intervalUsage = 0.65)
    {
        $this->intervalUsage = max(0.05, min($intervalUsage, 0.95));
        $this->connection = $connection;
    }

    /**
     * Enables the pessimistic mode.
     *
     * @param bool $pessimistic
     */
    public function setPessimistic($pessimistic)
    {
        $this->pessimistic = $pessimistic;
    }

    /**
     * Called whenever the server send data.
     *
     * @return void
     */
    protected function onServerActivity()
    {
        $this->checkDelayed();
    }

    /**
     * A frame with heartbeat details was detected.
     *
     * Class should set client or server interval.
     *
     * @param Frame $frame
     * @param array $beats
     * @return void
     */
    protected function onHeartbeatFrame(Frame $frame, array $beats)
    {
        if ($frame->getCommand() === self::FRAME_SERVER_CONNECTED) {
            $this->intervalServer = $beats[1];
            if ($this->intervalClient === null) {
                $this->intervalClient = $this->intervalServer;
            }
        } else {
            $this->intervalClient = $beats[0];
            $this->rememberActivity();
        }
    }

    /**
     * Must return the interval (ms) that should be used to detect a delay.
     *
     * @param integer $maximum
     * @return float
     */
    protected function calculateInterval($maximum)
    {
        $intervalUsed = $maximum * $this->intervalUsage;
        $this->assertReadTimeoutSufficient($intervalUsed);
        return $intervalUsed;
    }

    /**
     * Verify that the client configured heartbeats don't conflict with the connection read timeout.
     *
     * @param float $interval
     * @return void
     */
    private function assertReadTimeoutSufficient($interval)
    {
        $readTimeout = $this->connection->getReadTimeout();
        $readTimeoutMs = ($readTimeout[0] * 1000) + ($readTimeout[1] / 1000);

        if ($interval < $readTimeoutMs) {
            throw new HeartbeatException(
                'Client heartbeat is lower than connection read timeout, causing failing heartbeats.'
            );
        }
    }

    /**
     * Called whenever a activity is detected that was issued by the client.
     *
     * @return void
     */
    protected function onClientActivity()
    {
        $this->rememberActivity();
    }

    /**
     * @inheritdoc
     */
    protected function onPotentialConnectionStateActivity()
    {
        if ($this->pessimistic && $this->isEnabled()) {
            $this->onDelay();
        } else {
            $this->checkDelayed();
        }
    }

    /**
     * Send a beat to the server.
     *
     * @return void
     */
    protected function onDelay()
    {
        try {
            $this->connection->sendAlive($this->intervalClient / 1000);
        } catch (ConnectionException $e) {
            throw new HeartbeatException('Could not send heartbeat to server.', $e);
        }
        $this->rememberActivity();
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Network\Observer;

use Stomp\Network\Observer\Exception\HeartbeatException;
use Stomp\Transport\Frame;

/**
 * ServerAliveObserver an observer that checks for signals from server side.
 *
 * Use this to ensure that the server your listening to is still alive.
 *
 * If you want to signal the server that your client is still available check HeartbeatEmitter.
 *
 * @example $client->setHeartbeat(0, 2000); // indicate that we would receive server beats within a 2 second interval
 *          $client->getConnection()->getObservers()->addObserver(new ServerAliveObserver());
 *
 * @see HeartbeatEmitter
 * @package Stomp\Network\Observer
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ServerAliveObserver extends AbstractBeats
{
    /**
     * Defines the percentage amount of the calculated interval that will be used without emitting a beat.
     *
     * @var float
     */
    private $intervalUsage;

    /**
     * Emitter constructor.
     *
     * What is the interval usage?
     * The usage (percentage) defines the amount of agreed beat interval time,
     * that is allowed to pass before the observer decides that the server is delayed. (not alive anymore)
     *
     * A higher value increases the risk that a dead server is not detected over a given period.
     * A lower value increases the risk that a server is declared as dead when not.
     *
     * @param float $intervalUsage 150% default
     */
    public function __construct($intervalUsage = 1.5)
    {
        $this->intervalUsage = max(1, $intervalUsage);
    }

    /**
     * @inheritdoc
     */
    protected function onPotentialConnectionStateActivity()
    {
        $this->checkDelayed();
    }

    /**
     * @inheritdoc
     */
    protected function onServerActivity()
    {
        $this->rememberActivity();
    }

    /**
     * @inheritdoc
     */
    protected function onClientActivity()
    {
        // ignored here, as we see failures when the write fails
    }

    /**
     * @inheritdoc
     */
    protected function onDelay()
    {
        throw new HeartbeatException('The server failed to send expected heartbeats.');
    }

    /**
     * @inheritdoc
     */
    protected function onHeartbeatFrame(Frame $frame, array $beats)
    {
        if ($frame->getCommand() === self::FRAME_CLIENT_CONNECT) {
            $this->intervalClient = $beats[1];
        } else {
            $this->intervalServer = $beats[0];
            $this->rememberActivity();
        }
    }

    /**
     * @inheritdoc
     */
    protected function calculateInterval($maximum)
    {
        return $maximum * $this->intervalUsage;
    }
}


code:
<?php

/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Protocol;

use Stomp\Exception\StompException;
use Stomp\Transport\Frame;

/**
 * Stomp base protocol
 *
 *
 * @package Stomp
 * @author Hiram Chirino <hiram@hiramchirino.com>
 * @author Dejan Bosanac <dejan@nighttale.net>
 * @author Michael Caplan <mcaplan@labnet.net>
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Protocol
{
    /**
     * Client id used for durable subscriptions
     *
     * @var string
     */
    private $clientId;

    /**
     * @var string
     */
    private $version;

    /**
     * Server Version
     *
     * @var string
     */
    private $server;

    /**
     * Setup stomp protocol with configuration.
     *
     * @param string $clientId
     * @param string $version
     * @param string $server
     */
    public function __construct($clientId, $version = Version::VERSION_1_0, $server = null)
    {
        $this->clientId = $clientId;
        $this->server = $server;
        $this->version = $version;
    }

    /**
     * Get the connect frame
     *
     * @param string $login
     * @param string $passcode
     * @param array $versions
     * @param string $host
     * @param int[] $heartbeat
     * @return \Stomp\Transport\Frame
     */
    final public function getConnectFrame(
        $login = '',
        $passcode = '',
        array $versions = [],
        $host = null,
        $heartbeat = [0, 0]
    ) {
        $frame = $this->createFrame('CONNECT');
        $frame->legacyMode(true);

        if ($login || $passcode) {
            $frame->addHeaders(['login' => $login, 'passcode' => $passcode]);
        }

        if ($this->hasClientId()) {
            $frame['client-id'] = $this->getClientId();
        }

        if (!empty($versions)) {
            $frame['accept-version'] = implode(',', $versions);
        }

        $frame['host'] = $host;

        $frame['heart-beat'] = $heartbeat[0] . ',' . $heartbeat[1];

        return $frame;
    }

    /**
     * Get subscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param string $ack
     * @param string $selector
     * @return \Stomp\Transport\Frame
     * @throws StompException;
     */
    public function getSubscribeFrame($destination, $subscriptionId = null, $ack = 'auto', $selector = null)
    {
        // validate ACK types per spec
        // https://stomp.github.io/stomp-specification-1.0.html#frame-ACK
        // https://stomp.github.io/stomp-specification-1.1.html#ACK
        // https://stomp.github.io/stomp-specification-1.2.html#ACK
        if ($this->hasVersion(Version::VERSION_1_1)) {
            $validAcks = ['auto', 'client', 'client-individual'];
        } else {
            $validAcks = ['auto', 'client'];
        }
        if (!in_array($ack, $validAcks)) {
            throw new StompException(
                sprintf(
                    '"%s" is not a valid ack value for STOMP %s. A valid value is one of %s',
                    $ack,
                    $this->version,
                    implode(',', $validAcks)
                )
            );
        }

        $frame = $this->createFrame('SUBSCRIBE');

        $frame['destination'] = $destination;
        $frame['ack'] = $ack;
        $frame['id'] = $subscriptionId;
        $frame['selector'] = $selector;
        return $frame;
    }

    /**
     * Get unsubscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @return \Stomp\Transport\Frame
     */
    public function getUnsubscribeFrame($destination, $subscriptionId = null)
    {
        $frame = $this->createFrame('UNSUBSCRIBE');
        $frame['destination'] = $destination;
        $frame['id'] = $subscriptionId;
        return $frame;
    }

    /**
     * Get transaction begin frame.
     *
     * @param string $transactionId
     * @return \Stomp\Transport\Frame
     */
    public function getBeginFrame($transactionId = null)
    {
        $frame = $this->createFrame('BEGIN');
        $frame['transaction'] = $transactionId;
        return $frame;
    }

    /**
     * Get transaction commit frame.
     *
     * @param string $transactionId
     * @return \Stomp\Transport\Frame
     */
    public function getCommitFrame($transactionId = null)
    {
        $frame = $this->createFrame('COMMIT');
        $frame['transaction'] = $transactionId;
        return $frame;
    }

    /**
     * Get transaction abort frame.
     *
     * @param string $transactionId
     * @return \Stomp\Transport\Frame
     */
    public function getAbortFrame($transactionId = null)
    {
        $frame = $this->createFrame('ABORT');
        $frame['transaction'] = $transactionId;
        return $frame;
    }

    /**
     * Get message acknowledge frame.
     *
     * @param Frame $frame
     * @param string $transactionId
     * @return Frame
     */
    public function getAckFrame(Frame $frame, $transactionId = null)
    {
        $ack = $this->createFrame('ACK');
        $ack['transaction'] = $transactionId;
        if ($this->hasVersion(Version::VERSION_1_2)) {
            $ack['id'] = $frame->getMessageId();
        } else {
            $ack['message-id'] = $frame->getMessageId();
            if ($this->hasVersion(Version::VERSION_1_1)) {
                $ack['subscription'] = $frame['subscription'];
            }
        }
        return $ack;
    }

    /**
     * Get message not acknowledge frame.
     *
     * @param \Stomp\Transport\Frame $frame
     * @param string $transactionId
     * @param bool $requeue Requeue header
     * @return \Stomp\Transport\Frame
     * @throws StompException
     * @throws \LogicException
     */
    public function getNackFrame(Frame $frame, $transactionId = null, $requeue = null)
    {
        if ($requeue !== null) {
            throw new \LogicException('requeue header not supported');
        }
        if ($this->version === Version::VERSION_1_0) {
            throw new StompException('Stomp Version 1.0 has no support for NACK Frames.');
        }
        $nack = $this->createFrame('NACK');
        $nack['transaction'] = $transactionId;
        if ($this->hasVersion(Version::VERSION_1_2)) {
            $nack['id'] = $frame->getMessageId();
        } else {
            $nack['message-id'] = $frame->getMessageId();
            if ($this->hasVersion(Version::VERSION_1_1)) {
                $nack['subscription'] = $frame['subscription'];
            }
        }

        $nack['message-id'] = $frame->getMessageId();
        return $nack;
    }

    /**
     * Get the disconnect frame.
     *
     * @return \Stomp\Transport\Frame
     */
    public function getDisconnectFrame()
    {
        $frame = $this->createFrame('DISCONNECT');
        if ($this->hasClientId()) {
            $frame['client-id'] = $this->getClientId();
        }
        return $frame;
    }

    /**
     * Client Id is set
     *
     * @return boolean
     */
    public function hasClientId()
    {
        return (boolean) $this->clientId;
    }

    /**
     * Client Id is set
     *
     * @return string
     */
    public function getClientId()
    {
        return $this->clientId;
    }

    /**
     * Stomp Version
     *
     * @return string
     */
    public function getVersion()
    {
        return $this->version;
    }

    /**
     * Server Version Info
     *
     * @return string
     */
    public function getServer()
    {
        return $this->server;
    }

    /**
     * Checks if given version is included (equal or lower) in active protocol version.
     *
     * @param string $version
     * @return bool
     */
    public function hasVersion($version)
    {
        return version_compare($this->version, $version, '>=');
    }

    /**
     * Creates a Frame according to the detected STOMP version.
     *
     * @param string $command
     * @return Frame
     */
    protected function createFrame($command)
    {
        $frame = new Frame($command);

        if ($this->version === Version::VERSION_1_0) {
            $frame->legacyMode(true);
        }

        return $frame;
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\Protocol;

use Stomp\Broker\ActiveMq\ActiveMq;
use Stomp\Broker\Apollo\Apollo;
use Stomp\Broker\RabbitMq\RabbitMq;
use Stomp\Broker\OpenMq\OpenMq;
use Stomp\Exception\StompException;
use Stomp\Exception\UnexpectedResponseException;
use Stomp\Transport\Frame;

/**
 * Version determine stomp version and server dialect.
 *
 * @package Stomp\Protocol
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Version
{
    /**
     * Stomp Version 1.0
     */
    const VERSION_1_0 = '1.0';
    /**
     * Stomp Version 1.1
     */
    const VERSION_1_1 = '1.1';
    /**
     * Stomp Version 1.2
     */
    const VERSION_1_2 = '1.2';

    /**
     * @var Frame
     */
    private $frame;

    /**
     * Version constructor.
     *
     * @param Frame $frame
     * @throws StompException
     */
    public function __construct(Frame $frame)
    {
        if ($frame->getCommand() != 'CONNECTED') {
            throw new UnexpectedResponseException(
                $frame,
                sprintf('Expected a "CONNECTED" Frame to determine Version. Got a "%s" Frame!', $frame->getCommand())
            );
        }
        $this->frame = $frame;
    }

    /**
     * Returns the protocol to use.
     *
     * @param string $clientId
     * @param string $default server to use of no server detected
     * @return ActiveMq|Apollo|Protocol|RabbitMq
     */
    public function getProtocol($clientId, $default = 'ActiveMQ/5.11.1')
    {
        $server = trim($this->frame['server']) ?: $default;
        $version = $this->getVersion();
        if (stristr($server, 'rabbitmq') !== false) {
            return new RabbitMq($clientId, $version, $server);
        }
        if (stristr($server, 'apache-apollo') !== false) {
            return new Apollo($clientId, $version, $server);
        }
        if (stristr($server, 'activemq') !== false) {
            return new ActiveMq($clientId, $version, $server);
        }
        if (stristr($server, 'open message queue') !== false) {
            return new OpenMq($clientId, $version, $server);
        }
        return new Protocol($clientId, $version, $server);
    }

    /**
     * Detected version
     *
     * @return string
     */
    public function getVersion()
    {
        return $this->frame['version'] ?: self::VERSION_1_0;
    }

    /**
     * Check if version is same or newer than given one.
     *
     * @param string $version to check against
     * @return boolean
     */
    public function hasVersion($version)
    {
        return version_compare($this->getVersion(), $version, '>=');
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp;

use Stomp\Exception\StompException;
use Stomp\Protocol\Protocol;
use Stomp\Transport\Frame;
use Stomp\Transport\Message;

/**
 * Simple Stomp Client
 *
 * This is a legacy implementation of the old Stomp Client (Version 2-3).
 * It's an almost stateless client, only wrapping some protocol calls for you.
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class SimpleStomp
{
    /**
     * @var Client
     */
    protected $client;

    /**
     * LegacyStomp constructor.
     *
     * @param Client $client
     */
    public function __construct(Client $client)
    {
        $this->client = $client;
    }

    /**
     * Read response frame from server
     *
     * @return Frame|false when no frame to read
     */
    public function read()
    {
        return $this->client->readFrame();
    }

    /**
     * Register to listen to a given destination
     *
     * @param string $destination Destination queue
     * @param null $subscriptionId
     * @param string $ack
     * @param string $selector
     * @param array $header
     * @return bool
     */
    public function subscribe($destination, $subscriptionId = null, $ack = 'auto', $selector = null, array $header = [])
    {
        return $this->client->sendFrame(
            $this->getProtocol()->getSubscribeFrame($destination, $subscriptionId, $ack, $selector)->addHeaders($header)
        );
    }

    /**
     * @return Protocol
     */
    protected function getProtocol()
    {
        return $this->client->getProtocol();
    }

    /**
     * Send a message
     *
     * @param string $destination
     * @param Message $message
     * @return bool
     * @throws StompException
     */
    public function send($destination, Message $message)
    {
        return $this->client->send($destination, $message);
    }

    /**
     * Remove an existing subscription
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param array $header
     * @return boolean
     * @throws StompException
     */
    public function unsubscribe($destination, $subscriptionId = null, array $header = [])
    {
        return $this->client->sendFrame(
            $this->getProtocol()->getUnsubscribeFrame($destination, $subscriptionId)->addHeaders($header)
        );
    }

    /**
     * Start a transaction
     *
     * @param string $transactionId
     * @return boolean
     * @throws StompException
     */
    public function begin($transactionId = null)
    {
        return $this->client->sendFrame($this->getProtocol()->getBeginFrame($transactionId));
    }

    /**
     * Commit a transaction in progress
     *
     * @param string $transactionId
     * @return boolean
     * @throws StompException
     */
    public function commit($transactionId = null)
    {
        return $this->client->sendFrame($this->getProtocol()->getCommitFrame($transactionId));
    }

    /**
     * Roll back a transaction in progress
     *
     * @param string $transactionId
     * @return bool
     */
    public function abort($transactionId = null)
    {
        return $this->client->sendFrame($this->getProtocol()->getAbortFrame($transactionId));
    }

    /**
     * Acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @return void
     */
    public function ack(Frame $frame)
    {
        $this->client->sendFrame($this->getProtocol()->getAckFrame($frame), false);
    }

    /**
     * Not acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @return void
     */
    public function nack(Frame $frame)
    {
        $this->client->sendFrame($this->getProtocol()->getNackFrame($frame), false);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp;

use Stomp\States\IStateful;
use Stomp\States\Meta\SubscriptionList;
use Stomp\States\ProducerState;
use Stomp\States\StateSetter;
use Stomp\Transport\Frame;
use Stomp\Transport\Message;

/**
 * Stateful Stomp Client
 *
 * This is a stateful implementation of a stomp client.
 * This client will help you using stomp in a safe way by using the state machine pattern.
 *
 * @package Stomp
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class StatefulStomp extends StateSetter implements IStateful
{

    /**
     * active state
     *
     * @var IStateful
     */
    private $state;

    /**
     * @var Client
     */
    private $client;

    /**
     * StatefulStomp constructor.
     *
     * @param Client $client
     */
    public function __construct(Client $client)
    {
        $this->client = $client;
        $this->state = new ProducerState($client, $this);
    }

    /**
     * Acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @return void
     */
    public function ack(Frame $frame)
    {
        $this->state->ack($frame);
    }

    /**
     * Not acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @param bool $requeue requeue header not supported in all brokers
     * @return void
     */
    public function nack(Frame $frame, $requeue = null)
    {
        $this->state->nack($frame, $requeue);
    }

    /**
     * Send a message.
     *
     * @param string $destination
     * @param \Stomp\Transport\Message $message
     * @return bool
     */
    public function send($destination, Message $message)
    {
        return $this->state->send($destination, $message);
    }

    /**
     * Begins an transaction.
     *
     * @return void
     */
    public function begin()
    {
        $this->state->begin();
    }

    /**
     * Commit current transaction.
     *
     * @return void
     */
    public function commit()
    {
        $this->state->commit();
    }

    /**
     * Abort current transaction.
     *
     * @return void
     */
    public function abort()
    {
        $this->state->abort();
    }

    /**
     * Subscribe to given destination.
     *
     * Returns the subscriptionId used for this.
     *
     * @param string $destination
     * @param string $selector
     * @param string $ack
     * @param array $header
     * @return int
     */
    public function subscribe($destination, $selector = null, $ack = 'auto', array $header = [])
    {
        return $this->state->subscribe($destination, $selector, $ack, $header);
    }

    /**
     * Unsubscribe from current or given destination.
     *
     * @param int $subscriptionId
     * @return void
     */
    public function unsubscribe($subscriptionId = null)
    {
        $this->state->unsubscribe($subscriptionId);
    }

    /**
     * Returns as list of all active subscriptions.
     *
     * @return SubscriptionList
     */
    public function getSubscriptions()
    {
        return $this->state->getSubscriptions();
    }

    /**
     * Read a frame
     *
     * @return \Stomp\Transport\Frame|false
     */
    public function read()
    {
        return $this->state->read();
    }

    /**
     * Current State
     *
     * @return IStateful
     */
    public function getState()
    {
        return $this->state;
    }

    /**
     * Changes the current state.
     *
     * @param IStateful $state
     * @return mixed
     */
    protected function setState(IStateful $state)
    {
        $this->state = $state;
    }

    /**
     * Returns the used client.
     *
     * @return Client
     */
    public function getClient()
    {
        return $this->client;
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use InvalidArgumentException;
use Stomp\States\Meta\Subscription;
use Stomp\States\Meta\SubscriptionList;
use Stomp\Transport\Frame;
use Stomp\Transport\Message;
use Stomp\Util\IdGenerator;

/**
 * ConsumerState client acts as a consumer.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ConsumerState extends StateTemplate
{
    /**
     * Subscription ack mode
     *
     * @var string
     */
    protected $ack;
    /**
     * Subscription selector
     *
     * @var string
     */
    protected $selector;
    /**
     * Subscription target
     *
     * @var string
     */
    protected $destination;

    /**
     * SubscriptionId
     * @var int
     */
    protected $subId;

    /**
     * @var SubscriptionList
     */
    protected $subscriptions;

    /**
     * @inheritdoc
     */
    protected function init(array $options = [])
    {
        $this->subscriptions = new SubscriptionList();
        if (isset($options['subscriptions'])) {
            $this->subscriptions = $options['subscriptions'];
        } else {
            $this->subscribe($options['destination'], $options['selector'], $options['ack'], $options['header']);
        }
        return $this->subscriptions->getLast()->getSubscriptionId();
    }

    /**
     * @inheritdoc
     */
    public function ack(Frame $frame)
    {
        $this->getClient()->sendFrame($this->getProtocol()->getAckFrame($frame), false);
    }

    /**
     * @inheritdoc
     */
    public function nack(Frame $frame, $requeue = null)
    {
        $this->getClient()->sendFrame($this->getProtocol()->getNackFrame($frame, null, $requeue), false);
    }

    /**
     * @inheritdoc
     */
    public function send($destination, Message $message)
    {
        return $this->getClient()->send($destination, $message);
    }

    /**
     * @inheritdoc
     */
    public function begin()
    {
        $this->setState(new ConsumerTransactionState($this->getClient(), $this->getBase()), $this->getOptions());
    }

    /**
     * @inheritdoc
     */
    public function subscribe($destination, $selector, $ack, array $header = [])
    {
        $subscription = new Subscription($destination, $selector, $ack, IdGenerator::generateId(), $header);
        $this->getClient()->sendFrame(
            $this->getProtocol()->getSubscribeFrame(
                $subscription->getDestination(),
                $subscription->getSubscriptionId(),
                $subscription->getAck(),
                $subscription->getSelector()
            )->addHeaders($header)
        );
        $this->subscriptions[$subscription->getSubscriptionId()] = $subscription;
        return $subscription->getSubscriptionId();
    }

    /**
     * @inheritdoc
     */
    public function unsubscribe($subscriptionId = null)
    {
        if ($this->endSubscription($subscriptionId)) {
            $this->setState(
                new ProducerState($this->getClient(), $this->getBase())
            );
        }
    }

    /**
     * Closes given subscription or last opened.
     *
     * @param string $subscriptionId
     * @return bool true if last one was closed
     */
    protected function endSubscription($subscriptionId = null)
    {
        if (!$subscriptionId) {
            $subscriptionId = $this->subscriptions->getLast()->getSubscriptionId();
        }
        if (!isset($this->subscriptions[$subscriptionId])) {
            throw new InvalidArgumentException(sprintf('%s is no active subscription!', $subscriptionId));
        }
        $subscription = $this->subscriptions[$subscriptionId];
        $this->getClient()->sendFrame(
            $this->getProtocol()->getUnsubscribeFrame(
                $subscription->getDestination(),
                $subscription->getSubscriptionId()
            )
        );
        IdGenerator::releaseId($subscription->getSubscriptionId());

        unset($this->subscriptions[$subscription->getSubscriptionId()]);

        if ($this->subscriptions->count() == 0) {
            return true;
        }
        return false;
    }

    /**
     * @inheritdoc
     */
    public function read()
    {
        return $this->getClient()->readFrame();
    }

    /**
     * @inheritdoc
     */
    public function getSubscriptions()
    {
        return $this->subscriptions;
    }

    /**
     * @inheritdoc
     */
    protected function getOptions()
    {
        return [
            'subscriptions' => $this->subscriptions
        ];
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use Stomp\States\Exception\InvalidStateException;
use Stomp\Transport\Frame;

/**
 * ConsumerTransactionState client is a consumer within an transaction.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ConsumerTransactionState extends ConsumerState
{
    use TransactionsTrait;

    /**
     * @inheritdoc
     */
    protected function init(array $options = [])
    {
        $this->initTransaction($options);
        return parent::init($options);
    }

    /**
     * @inheritdoc
     */
    public function commit()
    {
        $this->getClient()->sendFrame(
            $this->getProtocol()->getCommitFrame($this->transactionId)
        );
        $this->setState(new ConsumerState($this->getClient(), $this->getBase()), parent::getOptions());
    }

    /**
     * @inheritdoc
     */
    public function abort()
    {
        $this->transactionAbort();
        $this->setState(new ConsumerState($this->getClient(), $this->getBase()), parent::getOptions());
    }

    /**
     * @inheritdoc
     */
    public function ack(Frame $frame)
    {
        $this->getClient()->sendFrame($this->getProtocol()->getAckFrame($frame, $this->transactionId), false);
    }

    /**
     * @inheritdoc
     */
    public function nack(Frame $frame, $requeue = null)
    {
        $this->getClient()->sendFrame(
            $this->getProtocol()->getNackFrame($frame, $this->transactionId, $requeue),
            false
        );
    }

    /**
     * @inheritdoc
     */
    public function unsubscribe($subscriptionId = null)
    {
        if ($this->endSubscription($subscriptionId)) {
            $this->setState(
                new ProducerTransactionState($this->getClient(), $this->getBase()),
                ['transactionId' => $this->transactionId]
            );
        }
    }

    /**
     * @inheritdoc
     */
    public function begin()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States\Exception;

use Stomp\Exception\StompException;
use Stomp\States\IStateful;

/**
 * InvalidStateException indicates that an call to an operation is not possible in current state.
 *
 * @package Stomp\States\Exception
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class InvalidStateException extends StompException
{

    /**
     * InvalidStateException constructor.
     *
     * @param IStateful $state
     * @param string $method
     */
    public function __construct(IStateful $state, $method)
    {
        parent::__construct(sprintf('"%s" is not allowed in "%s".', $method, get_class($state)));
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use Stomp\States\Meta\SubscriptionList;
use Stomp\Transport\Frame;
use Stomp\Transport\Message;

/**
 * Interface IStateful methods that must be treated in every stomp state.
 *
 * @package Stomp\States
 */
interface IStateful
{
    /**
     * Acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @return void
     */
    public function ack(Frame $frame);

    /**
     * Not acknowledge consumption of a message from a subscription
     *
     * @param Frame $frame
     * @param bool $requeue Requeue header not supported on all brokers
     * @return void
     */
    public function nack(Frame $frame, $requeue = null);

    /**
     * Send a message.
     *
     * @param string $destination
     * @param \Stomp\Transport\Message $message
     * @return bool
     */
    public function send($destination, Message $message);

    /**
     * Begins an transaction.
     *
     * @return void
     */
    public function begin();

    /**
     * Commit current transaction.
     *
     * @return void
     */
    public function commit();

    /**
     * Abort current transaction.
     *
     * @return void
     */
    public function abort();

    /**
     * Subscribe to given destination.
     *
     * Returns the subscriptionId used for this.
     *
     * @param string $destination
     * @param string $selector
     * @param string $ack
     * @param array  $header
     * @return int
     */
    public function subscribe($destination, $selector, $ack, array $header = []);

    /**
     * Unsubscribe from current or given destination.
     *
     * @param int $subscriptionId
     * @return void
     */
    public function unsubscribe($subscriptionId = null);

    /**
     * Read a frame
     *
     * @return \Stomp\Transport\Frame|false
     */
    public function read();

    /**
     * Returns as list of all active subscriptions.
     *
     * @return SubscriptionList
     */
    public function getSubscriptions();
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States\Meta;

use Stomp\Transport\Frame;

/**
 * Subscription Meta info
 *
 * @package Stomp\States\Meta
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Subscription
{
    /**
     * @var int
     */
    private $subscriptionId;

    /**
     * @var String
     */
    private $selector;

    /**
     * @var String
     */
    private $destination;

    /**
     * @var String
     */
    private $ack;

    /**
     * @var array
     */
    private $header;

    /**
     * Subscription constructor.
     * @param String $destination
     * @param String $selector
     * @param String $ack
     * @param int $subscriptionId
     * @param array $header additionally passed to create this subscription
     */
    public function __construct($destination, $selector, $ack, $subscriptionId, array $header = [])
    {
        $this->subscriptionId = $subscriptionId;
        $this->selector = $selector;
        $this->destination = $destination;
        $this->ack = $ack;
        $this->header = $header;
    }

    /**
     * @return int
     */
    public function getSubscriptionId()
    {
        return $this->subscriptionId;
    }

    /**
     * @return String
     */
    public function getSelector()
    {
        return $this->selector;
    }

    /**
     * @return String
     */
    public function getDestination()
    {
        return $this->destination;
    }

    /**
     * @return String
     */
    public function getAck()
    {
        return $this->ack;
    }

    /**
     * @return array
     */
    public function getHeader()
    {
        return $this->header;
    }

    /**
     * Checks if the given frame belongs to current Subscription.
     *
     * @param Frame $frame
     * @return bool
     */
    public function belongsTo(Frame $frame)
    {
        return ($frame['subscription'] == $this->subscriptionId);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States\Meta;

use ArrayAccess;
use ArrayIterator;
use Countable;
use IteratorAggregate;
use Stomp\Transport\Frame;

/**
 * SubscriptionList meta info for active subscriptions.
 *
 * @package Stomp\States\Meta
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class SubscriptionList implements IteratorAggregate, ArrayAccess, Countable
{

    /**
     * @var Subscription[]
     */
    private $subscriptions = [];

    /**
     * Returns the last added active Subscription.
     *
     * @return Subscription
     */
    public function getLast()
    {
        return end($this->subscriptions);
    }

    /**
     * Returns the subscription the frame belongs to or false if no matching subscription was found.
     *
     * @param Frame $frame
     * @return Subscription|false
     */
    public function getSubscription(Frame $frame)
    {
        foreach ($this->subscriptions as $subscription) {
            if ($subscription->belongsTo($frame)) {
                return $subscription;
            }
        }
        return false;
    }

    /**
     * @inheritdoc
     */
    public function getIterator()
    {
        return new ArrayIterator($this->subscriptions);
    }

    /**
     * @inheritdoc
     */
    public function offsetExists($offset)
    {
        return isset($this->subscriptions[$offset]);
    }

    /**
     * @inheritdoc
     */
    public function offsetGet($offset)
    {
        return $this->subscriptions[$offset];
    }

    /**
     * @inheritdoc
     */
    public function offsetSet($offset, $value)
    {
        $this->subscriptions[$offset] = $value;
    }

    /**
     * @inheritdoc
     */
    public function offsetUnset($offset)
    {
        unset($this->subscriptions[$offset]);
    }

    /**
     * @inheritdoc
     */
    public function count()
    {
        return count($this->subscriptions);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

/**
 * ProducerState client is working as a message producer.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ProducerState extends StateTemplate
{
    /**
     * @inheritdoc
     */
    protected function init(array $options = [])
    {
        // nothing to do here
    }

    /**
     * @inheritdoc
     */
    public function begin()
    {
        $this->setState(new ProducerTransactionState($this->getClient(), $this->getBase()));
    }

    /**
     * @inheritdoc
     */
    public function subscribe($destination, $selector, $ack, array $header = [])
    {
        return $this->setState(
            new ConsumerState($this->getClient(), $this->getBase()),
            ['destination' => $destination, 'selector' => $selector, 'ack' => $ack, 'header' => $header]
        );
    }

    /**
     * @inheritdoc
     */
    protected function getOptions()
    {
        return [];
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use Stomp\States\Exception\InvalidStateException;

/**
 * ProducerTransactionState client is working in an transaction scope as a message producer.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class ProducerTransactionState extends ProducerState
{
    use TransactionsTrait;

    /**
     * @inheritdoc
     */
    protected function init(array $options = [])
    {
        $this->initTransaction($options);
        parent::init($options);
    }

    /**
     * @inheritdoc
     */
    public function commit()
    {
        $this->transactionCommit();
        $this->setState(new ProducerState($this->getClient(), $this->getBase()), parent::getOptions());
    }

    /**
     * @inheritdoc
     */
    public function abort()
    {
        $this->transactionAbort();
        $this->setState(new ProducerState($this->getClient(), $this->getBase()), parent::getOptions());
    }

    /**
     * @inheritdoc
     */
    public function subscribe($destination, $selector, $ack, array $header = [])
    {
        return $this->setState(
            new ConsumerTransactionState($this->getClient(), $this->getBase()),
            $this->getOptions() +
            [
                'destination' => $destination,
                'selector' => $selector,
                'ack' => $ack,
                'header' => $header
            ]
        );
    }

    /**
     * @inheritdoc
     */
    public function begin()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

/**
 * StateSetter allows to change a state.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
abstract class StateSetter
{
    /**
     * Change current state to given one, might return creation data from state.
     *
     * @param IStateful $state
     * @return mixed
     */
    abstract protected function setState(IStateful $state);
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use Stomp\Client;
use Stomp\Protocol\Protocol;
use Stomp\StatefulStomp;
use Stomp\States\Exception\InvalidStateException;
use Stomp\States\Meta\SubscriptionList;
use Stomp\Transport\Frame;
use Stomp\Transport\Message;

/**
 * StateTemplate for StompStates.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
abstract class StateTemplate extends StateSetter implements IStateful
{
    /**
     * @var Client
     */
    private $client;

    /**
     * StateMachine
     *
     * @var StatefulStomp
     */
    private $base;

    /**
     * StateTemplate constructor.
     * @param Client $client
     * @param StatefulStomp $base
     */
    public function __construct(Client $client, StatefulStomp $base)
    {
        $this->client = $client;
        $this->base = $base;
    }

    /**
     * Returns the base StateMachine.
     *
     * @return StatefulStomp
     */
    protected function getBase()
    {
        return $this->base;
    }

    /**
     * Activates the current state, after it has been applied on base.
     *
     * @param array $options
     * @return mixed
     */
    abstract protected function init(array $options = []);

    /**
     * Returns the options needed in current state.
     *
     * @return array
     */
    abstract protected function getOptions();

    /**
     * @return Client
     */
    protected function getClient()
    {
        return $this->client;
    }

    /**
     * @return Protocol
     */
    protected function getProtocol()
    {
        return $this->client->getProtocol();
    }

    /**
     * @inheritdoc
     */
    protected function setState(IStateful $state, array $options = [])
    {
        $init = null;
        if ($state instanceof StateTemplate) {
            $init = $state->init($options);
        }
        $this->base->setState($state);
        return $init;
    }

    /**
     * @inheritdoc
     */
    public function ack(Frame $frame)
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function nack(Frame $frame, $requeue = null)
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function send($destination, Message $message)
    {
        return $this->getClient()->send($destination, $message);
    }

    /**
     * @inheritdoc
     */
    public function begin()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function commit()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function abort()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function subscribe($destination, $selector, $ack, array $header = [])
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function unsubscribe($subscriptionId = null)
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function read()
    {
        throw new InvalidStateException($this, __FUNCTION__);
    }

    /**
     * @inheritdoc
     */
    public function getSubscriptions()
    {
        return new SubscriptionList();
    }
}


code:
<?php
/*
 * This file is part of the Stomp package.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Stomp\States;

use Stomp\Client;
use Stomp\Protocol\Protocol;
use Stomp\Transport\Message;
use Stomp\Util\IdGenerator;

/**
 * TransactionsTrait provides base logic for all transaction based states.
 *
 * @package Stomp\States
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
trait TransactionsTrait
{

    /**
     * @return Protocol
     */
    abstract public function getProtocol();

    /**
     * @return Client
     */
    abstract public function getClient();

    /**
     * Id used for current transaction.
     *
     * @var int|string
     */
    protected $transactionId;

    /**
     * Init the transaction state.
     *
     * @param array $options
     */
    protected function initTransaction(array $options = [])
    {
        if (!isset($options['transactionId'])) {
            $this->transactionId = IdGenerator::generateId();
            $this->getClient()->sendFrame(
                $this->getProtocol()->getBeginFrame($this->transactionId)
            );
        } else {
            $this->transactionId = $options['transactionId'];
        }
    }

    /**
     * Options for this transaction state.
     *
     * @return array
     */
    protected function getOptions()
    {
        return ['transactionId' => $this->transactionId];
    }

    /**
     * Send a message within this transaction.
     *
     * @param string $destination
     * @param \Stomp\Transport\Message $message
     * @return bool
     */
    public function send($destination, Message $message)
    {
        return $this->getClient()->send($destination, $message, ['transaction' => $this->transactionId], false);
    }

    /**
     * Commit current transaction.
     */
    protected function transactionCommit()
    {
        $this->getClient()->sendFrame($this->getProtocol()->getCommitFrame($this->transactionId));
        IdGenerator::releaseId($this->transactionId);
    }

    /**
     * Abort the current transaction.
     */
    protected function transactionAbort()
    {
        $this->getClient()->sendFrame($this->getProtocol()->getAbortFrame($this->transactionId));
        IdGenerator::releaseId($this->transactionId);
    }
}