code:
<?php

/**
 * Pure-PHP ssh-agent client.
 *
 * PHP version 5
 *
 * Here are some examples of how to use this library:
 * <code>
 * <?php
 *    include 'vendor/autoload.php';
 *
 *    $agent = new \phpseclib\System\SSH\Agent();
 *
 *    $ssh = new \phpseclib\Net\SSH2('www.domain.tld');
 *    if (!$ssh->login('username', $agent)) {
 *        exit('Login Failed');
 *    }
 *
 *    echo $ssh->exec('pwd');
 *    echo $ssh->exec('ls -la');
 * ?>
 * </code>
 *
 * @category  System
 * @package   SSH\Agent
 * @author    Jim Wigginton <terrafrost@php.net>
 * @copyright 2014 Jim Wigginton
 * @license   http://www.opensource.org/licenses/mit-license.html  MIT License
 * @link      http://phpseclib.sourceforge.net
 * @internal  See http://api.libssh.org/rfc/PROTOCOL.agent
 */

namespace phpseclib\System\SSH;

use ParagonIE\ConstantTime\Base64;
use phpseclib\Crypt\RSA;
use phpseclib\Exception\BadConfigurationException;
use phpseclib\System\SSH\Agent\Identity;

/**
 * Pure-PHP ssh-agent client identity factory
 *
 * requestIdentities() method pumps out \phpseclib\System\SSH\Agent\Identity objects
 *
 * @package SSH\Agent
 * @author  Jim Wigginton <terrafrost@php.net>
 * @access  internal
 */
class Agent
{
    /**#@+
     * Message numbers
     *
     * @access private
     */
    // to request SSH1 keys you have to use SSH_AGENTC_REQUEST_RSA_IDENTITIES (1)
    const SSH_AGENTC_REQUEST_IDENTITIES = 11;
    // this is the SSH2 response; the SSH1 response is SSH_AGENT_RSA_IDENTITIES_ANSWER (2).
    const SSH_AGENT_IDENTITIES_ANSWER = 12;
    // the SSH1 request is SSH_AGENTC_RSA_CHALLENGE (3)
    const SSH_AGENTC_SIGN_REQUEST = 13;
    // the SSH1 response is SSH_AGENT_RSA_RESPONSE (4)
    const SSH_AGENT_SIGN_RESPONSE = 14;
    /**#@-*/

    /**@+
     * Agent forwarding status
     *
     * @access private
     */
    // no forwarding requested and not active
    const FORWARD_NONE = 0;
    // request agent forwarding when opportune
    const FORWARD_REQUEST = 1;
    // forwarding has been request and is active
    const FORWARD_ACTIVE = 2;
    /**#@-*/

    /**
     * Unused
     */
    const SSH_AGENT_FAILURE = 5;

    /**
     * Socket Resource
     *
     * @var resource
     * @access private
     */
    var $fsock;

    /**
     * Agent forwarding status
     *
     * @access private
     */
    var $forward_status = self::FORWARD_NONE;

    /**
     * Buffer for accumulating forwarded authentication
     * agent data arriving on SSH data channel destined
     * for agent unix socket
     *
     * @access private
     */
    var $socket_buffer = '';

    /**
     * Tracking the number of bytes we are expecting
     * to arrive for the agent socket on the SSH data
     * channel
     */
    var $expected_bytes = 0;

    /**
     * Default Constructor
     *
     * @return \phpseclib\System\SSH\Agent
     * @throws \phpseclib\Exception\BadConfigurationException if SSH_AUTH_SOCK cannot be found
     * @throws \RuntimeException on connection errors
     * @access public
     */
    function __construct()
    {
        switch (true) {
            case isset($_SERVER['SSH_AUTH_SOCK']):
                $address = $_SERVER['SSH_AUTH_SOCK'];
                break;
            case isset($_ENV['SSH_AUTH_SOCK']):
                $address = $_ENV['SSH_AUTH_SOCK'];
                break;
            default:
                throw new BadConfigurationException('SSH_AUTH_SOCK not found');
        }

        $this->fsock = fsockopen('unix://' . $address, 0, $errno, $errstr);
        if (!$this->fsock) {
            throw new \RuntimeException("Unable to connect to ssh-agent (Error $errno: $errstr)");
        }
    }

    /**
     * Request Identities
     *
     * See "2.5.2 Requesting a list of protocol 2 keys"
     * Returns an array containing zero or more \phpseclib\System\SSH\Agent\Identity objects
     *
     * @return array
     * @throws \RuntimeException on receipt of unexpected packets
     * @access public
     */
    function requestIdentities()
    {
        if (!$this->fsock) {
            return array();
        }

        $packet = pack('NC', 1, self::SSH_AGENTC_REQUEST_IDENTITIES);
        if (strlen($packet) != fputs($this->fsock, $packet)) {
            throw new \RuntimeException('Connection closed while requesting identities');
        }

        $length = current(unpack('N', fread($this->fsock, 4)));
        $type = ord(fread($this->fsock, 1));
        if ($type != self::SSH_AGENT_IDENTITIES_ANSWER) {
            throw new \RuntimeException('Unable to request identities');
        }

        $identities = array();
        $keyCount = current(unpack('N', fread($this->fsock, 4)));
        for ($i = 0; $i < $keyCount; $i++) {
            $length = current(unpack('N', fread($this->fsock, 4)));
            $key_blob = fread($this->fsock, $length);
            $key_str = 'ssh-rsa ' . Base64::encode($key_blob);
            $length = current(unpack('N', fread($this->fsock, 4)));
            if ($length) {
                $key_str.= ' ' . fread($this->fsock, $length);
            }
            $length = current(unpack('N', substr($key_blob, 0, 4)));
            $key_type = substr($key_blob, 4, $length);
            switch ($key_type) {
                case 'ssh-rsa':
                    $key = new RSA();
                    $key->load($key_str);
                    break;
                case 'ssh-dss':
                    // not currently supported
                    break;
            }
            // resources are passed by reference by default
            if (isset($key)) {
                $identity = new Identity($this->fsock);
                $identity->setPublicKey($key);
                $identity->setPublicKeyBlob($key_blob);
                $identities[] = $identity;
                unset($key);
            }
        }

        return $identities;
    }

    /**
     * Signal that agent forwarding should
     * be requested when a channel is opened
     *
     * @param Net_SSH2 $ssh
     * @return bool
     * @access public
     */
    function startSSHForwarding($ssh)
    {
        if ($this->forward_status == self::FORWARD_NONE) {
            $this->forward_status = self::FORWARD_REQUEST;
        }
    }

    /**
     * Request agent forwarding of remote server
     *
     * @param Net_SSH2 $ssh
     * @return bool
     * @access private
     */
    function _request_forwarding($ssh)
    {
        $request_channel = $ssh->_get_open_channel();
        if ($request_channel === false) {
            return false;
        }

        $packet = pack(
            'CNNa*C',
            NET_SSH2_MSG_CHANNEL_REQUEST,
            $ssh->server_channels[$request_channel],
            strlen('auth-agent-req@openssh.com'),
            'auth-agent-req@openssh.com',
            1
        );

        $ssh->channel_status[$request_channel] = NET_SSH2_MSG_CHANNEL_REQUEST;

        if (!$ssh->_send_binary_packet($packet)) {
            return false;
        }

        $response = $ssh->_get_channel_packet($request_channel);
        if ($response === false) {
            return false;
        }

        $ssh->channel_status[$request_channel] = NET_SSH2_MSG_CHANNEL_OPEN;
        $this->forward_status = self::FORWARD_ACTIVE;

        return true;
    }

    /**
     * On successful channel open
     *
     * This method is called upon successful channel
     * open to give the SSH Agent an opportunity
     * to take further action. i.e. request agent forwarding
     *
     * @param Net_SSH2 $ssh
     * @access private
     */
    function _on_channel_open($ssh)
    {
        if ($this->forward_status == self::FORWARD_REQUEST) {
            $this->_request_forwarding($ssh);
        }
    }

    /**
     * Forward data to SSH Agent and return data reply
     *
     * @param string $data
     * @return data from SSH Agent
     * @throws \RuntimeException on connection errors
     * @access private
     */
    function _forward_data($data)
    {
        if ($this->expected_bytes > 0) {
            $this->socket_buffer.= $data;
            $this->expected_bytes -= strlen($data);
        } else {
            $agent_data_bytes = current(unpack('N', $data));
            $current_data_bytes = strlen($data);
            $this->socket_buffer = $data;
            if ($current_data_bytes != $agent_data_bytes + 4) {
                $this->expected_bytes = ($agent_data_bytes + 4) - $current_data_bytes;
                return false;
            }
        }

        if (strlen($this->socket_buffer) != fwrite($this->fsock, $this->socket_buffer)) {
            throw new \RuntimeException('Connection closed attempting to forward data to SSH agent');
        }

        $this->socket_buffer = '';
        $this->expected_bytes = 0;

        $agent_reply_bytes = current(unpack('N', fread($this->fsock, 4)));

        $agent_reply_data = fread($this->fsock, $agent_reply_bytes);
        $agent_reply_data = current(unpack('a*', $agent_reply_data));

        return pack('Na*', $agent_reply_bytes, $agent_reply_data);
    }
}

repo_name: andremichelle/gnu

license: mit


code:
<?php
/**
 * Pure-PHP ssh-agent client.
 *
 * PHP version 5
 *
 * @category  System
 * @package   SSH\Agent
 * @author    Jim Wigginton <terrafrost@php.net>
 * @copyright 2009 Jim Wigginton
 * @license   http://www.opensource.org/licenses/mit-license.html  MIT License
 * @link      http://phpseclib.sourceforge.net
 * @internal  See http://api.libssh.org/rfc/PROTOCOL.agent
 */

namespace phpseclib\System\SSH\Agent;

use phpseclib\Crypt\RSA;
use phpseclib\Exception\UnsupportedAlgorithmException;
use phpseclib\System\SSH\Agent;

/**
 * Pure-PHP ssh-agent client identity object
 *
 * Instantiation should only be performed by \phpseclib\System\SSH\Agent class.
 * This could be thought of as implementing an interface that phpseclib\Crypt\RSA
 * implements. ie. maybe a Net_SSH_Auth_PublicKey interface or something.
 * The methods in this interface would be getPublicKey and sign since those are the
 * methods phpseclib looks for to perform public key authentication.
 *
 * @package SSH\Agent
 * @author  Jim Wigginton <terrafrost@php.net>
 * @access  internal
 */
class Identity
{
    /**
     * Key Object
     *
     * @var \phpseclib\Crypt\RSA
     * @access private
     * @see self::getPublicKey()
     */
    var $key;

    /**
     * Key Blob
     *
     * @var string
     * @access private
     * @see self::sign()
     */
    var $key_blob;

    /**
     * Socket Resource
     *
     * @var resource
     * @access private
     * @see self::sign()
     */
    var $fsock;

    /**
     * Default Constructor.
     *
     * @param resource $fsock
     * @return \phpseclib\System\SSH\Agent\Identity
     * @access private
     */
    function __construct($fsock)
    {
        $this->fsock = $fsock;
    }

    /**
     * Set Public Key
     *
     * Called by \phpseclib\System\SSH\Agent::requestIdentities()
     *
     * @param \phpseclib\Crypt\RSA $key
     * @access private
     */
    function setPublicKey($key)
    {
        $this->key = $key;
        $this->key->setPublicKey();
    }

    /**
     * Set Public Key
     *
     * Called by \phpseclib\System\SSH\Agent::requestIdentities(). The key blob could be extracted from $this->key
     * but this saves a small amount of computation.
     *
     * @param string $key_blob
     * @access private
     */
    function setPublicKeyBlob($key_blob)
    {
        $this->key_blob = $key_blob;
    }

    /**
     * Get Public Key
     *
     * Wrapper for $this->key->getPublicKey()
     *
     * @param int $type optional
     * @return mixed
     * @access public
     */
    function getPublicKey($type = 'PKCS8')
    {
        return $this->key->getPublicKey($type);
    }

    /**
     * Sets the hash
     *
     * ssh-agent only supports signatures with sha1 hashes but to maintain BC with RSA.php this function exists
     *
     * @param string $hash optional
     * @throws \phpseclib\Exception\UnsupportedAlgorithmException if the algorithm is unsupported
     * @access public
     */
    function setHash($hash = 'sha1')
    {
        if ($hash != 'sha1') {
            throw new UnsupportedAlgorithmException('ssh-agent can only be used with the sha1 hash');
        }
    }

    /**
     * Create a signature
     *
     * See "2.6.2 Protocol 2 private key signature request"
     *
     * @param string $message
     * @param int $padding optional
     * @return string
     * @throws \RuntimeException on connection errors
     * @throws \phpseclib\Exception\UnsupportedAlgorithmException if the algorithm is unsupported
     * @access public
     */
    function sign($message, $padding = RSA::PADDING_PKCS1)
    {
        if ($padding != RSA::PADDING_PKCS1 && $padding != RSA::PADDING_RELAXED_PKCS1) {
            throw new UnsupportedAlgorithmException('ssh-agent can only create PKCS1 signatures');
        }

        // the last parameter (currently 0) is for flags and ssh-agent only defines one flag (for ssh-dss): SSH_AGENT_OLD_SIGNATURE
        $packet = pack('CNa*Na*N', Agent::SSH_AGENTC_SIGN_REQUEST, strlen($this->key_blob), $this->key_blob, strlen($message), $message, 0);
        $packet = pack('Na*', strlen($packet), $packet);
        if (strlen($packet) != fputs($this->fsock, $packet)) {
            throw new \RuntimeException('Connection closed during signing');
        }

        $length = current(unpack('N', fread($this->fsock, 4)));
        $type = ord(fread($this->fsock, 1));
        if ($type != Agent::SSH_AGENT_SIGN_RESPONSE) {
            throw new \RuntimeException('Unable to retreive signature');
        }

        $signature_blob = fread($this->fsock, $length - 1);
        // the only other signature format defined - ssh-dss - is the same length as ssh-rsa
        // the + 12 is for the other various SSH added length fields
        return substr($signature_blob, strlen('ssh-rsa') + 12);
    }
}


code:
<?php

/**
 * Bootstrapping File for phpseclib
 *
 * composer isn't a requirement for phpseclib 2.0 but this file isn't really required
 * either. it's a bonus for those using composer but if you're not phpseclib will
 * still work
 *
 * @license http://www.opensource.org/licenses/mit-license.html MIT License
 */
if (extension_loaded('mbstring')) {
    // 2 - MB_OVERLOAD_STRING
    if (ini_get('mbstring.func_overload') & 2) {
        throw new UnexpectedValueException(
            'Overloading of string functions using mbstring.func_overload ' .
            'is not supported by phpseclib.'
        );
    }
}


code:
# minimalist openssl.cnf file for use with phpseclib

HOME			= .
RANDFILE		= $ENV::HOME/.rnd

[ v3_ca ]


code:
travisci/ export-ignore
tests/ export-ignore


code:
nbproject
composer.lock
composer.phar
vendor
/.idea
/.project
/travisci/tmp
stomp-php.iml


code:
language: php

sudo: true

services:
  - docker

jobs:
  include:
    - &TEST
      php: "5.6"
      before_script:
        - ./travisci/bin/start.sh
        - composer install
        - ./travisci/bin/logs.sh

      script: vendor/bin/phpunit --debug

      after_script:
        - ./travisci/bin/stop.sh
    - <<: *TEST
      php: "7.0"
    - <<: *TEST
      php: "7.1"
    - <<: *TEST
      php: "7.2"
    - <<: *TEST
      php: "7.3"

    - stage: Static analysis with phpstan
      php: "7.3"
      script:
        - composer require --dev phpstan/phpstan
        - vendor/bin/phpstan analyse


code:
Changelog stomp-php
-------------------

4.0.0 (Full Stomp 1.1 / 1.2 Support)
------------------------------------

- Updated functional testsuite for different brokers (amq,aplo,rabbit), update travis-ci.
- Update Parser in order to be compliant with stomp-1.2 (https://stomp.github.io/stomp-specification-1.2.html)
- ACK Mode is now `auto` by default.
- Implement StateMachine Pattern `StatefulStomp`.
- Restructure project, new Namespaces `Protocol`, `Transport`, `Network`.
- Make Client much smaller, remove all non low level methods to `SimpleStomp`.
- Testsuite rework...
- Move examples to https://github.com/stomp-php/stomp-php-examples.
- add utils for `ActiveMq` durable subscription and `Apollo` queue browser.

4.0.1
-----
- do not throw read exception if the next byte after a complete read is a zero byte. (https://github.com/stomp-php/stomp-php/issues/39)
- allow to reset the parser internal state and buffer. (https://github.com/stomp-php/stomp-php/issues/40)

4.1.0
-----
- fixed header escaping in legacy mode in Parser (28ed8b6)
- when the Parser is working in legacyMode it should produce Frames working in legacyMode (d257b98)
- validate given ack SUBSCRIBE frame values against the STOMP spec (8043970)
- Added a OpenMq Broker (0638958)
- updated AMQ ci test version from 5.13.0 to 5.13.3 (ee37df6)
- updated RMQ ci test version from 3.5.6 to 3.6.2 (ee37df6)
- updated durable tests for RMQ (pass 'auto-delete' header and replaced 'persistent' with 'durable' header) (ee37df6)
- updated erlang from R14B04 to latest version (travis) (ee37df6)
- increase read timeout for amq abort test (ee37df6)

4.1.1
-----
- fixed Null-Byte handling, thanks to @andrewbelcher (https://github.com/stomp-php/stomp-php/issues/54)

4.1.2
-----
- update nack and ack Frames according to stomp 1.1 specification, thanks to @rawkode (https://github.com/stomp-php/stomp-php/issues/56)

4.1.3
-----
- persistent connection flag for socket, thanks to @tenerd (https://github.com/stomp-php/stomp-php/issues/57)

4.1.4
-----
- add headers getter in Frame class, thanks to @surrealcristian (https://github.com/stomp-php/stomp-php/issues/61)

4.1.5
-----
- fix for false exceptions in combination with pcntl_signal (https://github.com/stomp-php/stomp-php/pull/65)

4.2.0
-----
- fix invalid class hierarchy for `Stomp\Transport\Map` changed parent from `Frame` to `Message`
- add option to register own message type handlers in `Stomp\Transport\FrameFactory` (https://github.com/stomp-php/stomp-php/pull/64)

4.2.1
-----
- Update Connector to support `_`  in the broker uri, thanks to @campru (https://github.com/stomp-php/stomp-php/pull/66)

4.2.2
-----
- add support for outgoing heartbeats and a basic heartbeat emitter, than can be added to the connection (see stomp-php-examples) thanks to @andrewbelcher and @staabm (https://github.com/stomp-php/stomp-php/pull/72, https://github.com/stomp-php/stomp-php/pull/60)

4.3.0
------
- bug: Message incorrectly parsed when body contains \r\n\r\n but delimiter is \n\n thanks to @mattford (https://github.com/stomp-php/stomp-php/issues/93, https://github.com/stomp-php/stomp-php/issues/94)
- bc: #87 drop php-5.5 support (if you still need to use php-5.5 we can provide a release 4.2.* release for #93 too, but ask for it)
- #91, #90, #89, #88, #87, #86, #84, #83, #82, #81, #80, #78, #77, #76, #75, #74, #73 many code style improvements big thanks to @localheinz
Thanks to @staabm for all the reviews :)

4.3.1
------
- add requeue support on NACK for rabbitmq 4.3, thanks to @gm-ghanover (https://github.com/stomp-php/stomp-php/pull/102)

4.4.0
------
- change to non blocking mode. Add writeTimeout see https://github.com/stomp-php/stomp-php/blob/master/src/Network/Connection.php#L219 defaults to 3 seconds (#104 @Shaa1 thanks for reporting, https://github.com/stomp-php/stomp-php/pull/107)
- add git-export-ignore thanks to @staabm (https://github.com/stomp-php/stomp-php/pull/106)

4.4.1
------
- fix parser fails on header values that contain \r (thanks to @riven8192 for the fix, https://github.com/stomp-php/stomp-php/pull/110)

4.4.2
------
- fix client stuck at large messages (thanks to @ganeko for fix, https://github.com/stomp-php/stomp-php/pull/111)

4.4.3
------
- add configuration for connection read/write may bytes (thanks to @ganeko for this enhancement, https://github.com/stomp-php/stomp-php/pull/113)

4.5.0
------
- refined broken connection detection, not longer failing when `fread` returns empty string (thanks to @mlamm for reporting this, https://github.com/stomp-php/stomp-php/issues/115 and https://github.com/stomp-php/stomp-php/issues/114
- Clients should now use either `ServerAliveObserver` or `HeartbeatEmitter` to verify the state of the connection, otherwise a broken connection might not be detected - especially when the socket is based on ssl.
- stabilize client heartbeat implementation, added awareness for failing `Connection::sendAlive` calls, now causing `HeartbeatException`
- added `ServerAliveObserver` in order to simplify detection of dead connections.
- fixed connection read blocks on os signals until read timeout (thanks to @edefimov, https://github.com/stomp-php/stomp-php/issues/117)


code:
# Stomp PHP

[![Build Status](https://travis-ci.org/stomp-php/stomp-php.svg?branch=master)](https://travis-ci.org/stomp-php/stomp-php)

This project is a PHP [Stomp](http://stomp.github.com) Client that besides it implements the Stomp protocol fully,
brings some ActiveMQ and Apollo specific utils that could make your messaging from PHP easier.

## Credits

This library was initially developed by [Dejan Bosanac](https://github.com/dejanb).
We would like to thank you for your work and we're happy to continue it.

## Version choice

- For new projects you should use version `4.*` which requires `php-5.6`. Support for `php-5.6` ends with version `5.*`.
- For projects running older php versions you can use version `4.2.*` for `php-5.5` and `3.*` for `php-5.3`, please consider to update php.
- For running projects with `fusesource/stomp-php@2.x` clients you can use version `2.2.2`.
- All version newer that `2.x` won't be compatible with `fusesource/stomp-php`. (https://github.com/dejanb/stomp-php.)

## Installing

```bash
composer require stomp-php/stomp-php
```

## Examples

You find different usage tutorials in our example project https://github.com/stomp-php/stomp-php-examples.

### Connection Probing

It's hard to find out if a socket connection is still working or not, Stomp allows us to use heartbeats to test if client
and server are ready to serve messages.

You should use `\Stomp\Network\Observer\ServerAliveObserver` or `\Stomp\Network\Observer\HeartbeatEmitter` to receive or
send heartbeats. Doing so will ensure that your client will detect a broken connection in time. Please have a look at
https://github.com/stomp-php/stomp-php-examples for some example code with additonal comments.

## Replace fusesource/stomp-php

If you used `fusesource/stomp-php` before, you can use our `2.x` versions.

```json
    "require": {
        "stomp-php/stomp-php": "2.*"
    }
```

## Contributing

We code in `PSR2`, please use our predefined `pre_commit.sh` hook.

## Tests

To run the tests you first need to fetch the dependencies for the test suite
via composer:

    $ php composer.phar install

The functional testsuite is divided into three broker versions.
Currently it's running on `ActiveMq` (Port 61010), `Apollo` (61020), `RabbitMq` (61030).
Apollo should be configured to use admin:password and RabbitMq to guest:guest.
While ActiveMq must be configured to use no login at all.

You can setup all brokers by running `travisci/bin/start.sh`. Stop them by `travisci/bin/stop.sh`. (Docker is required.)

If you only like to run the functional generic tests, ensure Apollo is configured.
A basic setup can be achieved by running `./travisci/bin/apollo-mq.sh 1.7.1`.
(If you want to create a local running broker, you find the config / setup at `travisci/docker/apollo-mq/`)

## Licence

[Apache License Version 2.0](http://www.apache.org/licenses/LICENSE-2.0)


stomp-php/stomp-php

description: stomp support for PHP

keywords: messaging stomp jms activemq rabbitmq apollomq

Stomp\: src/

Stomp\Tests\: tests/


code:
parameters:
    level: 4
    paths:
        - src/


code:
<?xml version="1.0" encoding="UTF-8"?>

<!-- http://www.phpunit.de/manual/current/en/appendixes.configuration.html -->
<phpunit
    backupGlobals               = "false"
    backupStaticAttributes      = "false"
    colors                      = "true"
    convertErrorsToExceptions   = "true"
    convertNoticesToExceptions  = "true"
    convertWarningsToExceptions = "true"
    processIsolation            = "false"
    stopOnFailure               = "false"
    syntaxCheck                 = "false"
    bootstrap                   = "tests/bootstrap.php" >
    <filter>
        <blacklist>
            <directory>vendor</directory>
        </blacklist>
        <whitelist>
            <directory>src</directory>
        </whitelist>
    </filter>
    <testsuites>
        <testsuite name="stomp-php Functional Test Suite">
            <directory>tests/Functional/</directory>
        </testsuite>
        <testsuite name="stomp-php Special Test Cases Suite">
            <directory>tests/Cases/</directory>
        </testsuite>
        <testsuite name="stomp-php Unit Test Suite">
            <directory>tests/Unit/</directory>
        </testsuite>
    </testsuites>
    <logging>
        <log type="coverage-text" target="php://stdout" showUncoveredFiles="true"/>
    </logging>
</phpunit>


code:
#!/usr/bin/env bash

TEXT_RESET="\e[0m"
TEXT_BOLD="\e[1;30m"
TEXT_RED="\e[1;31m"
TEXT_GREEN="\e[1;32m"

function shc {
    printf "${TEXT_BOLD}${1}${TEXT_RESET}"
    sh $2

    if [ $? -ne 0 ]; then
        printf "${TEXT_RED}FAIL${TEXT_RESET}\n"
        exit 1
    else
        printf "${TEXT_GREEN}OK${TEXT_RESET}\n"
    fi
}

shc "PHPCS..." run_phpcs.sh
shc "PHPUnit...\n" run_phpunit.sh

# run_phpcpd.sh


code:
#!/usr/bin/env bash

./vendor/bin/phpcs --extensions=php --standard=PSR2 --ignore=vendor src tests


code:
#!/usr/bin/env bash

./vendor/bin/phpunit --no-configuration tests/Unit/Stomp


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\ActiveMq;

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

/**
 * ActiveMq 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 ActiveMq extends Protocol
{
    /**
     * Prefetch Size for subscriptions.
     *
     * @var int
     */
    private $prefetchSize = 1;

    /**
     * ActiveMq 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['activemq.prefetchSize'] = $this->prefetchSize;
        if ($durable) {
            $frame['activemq.subscriptionName'] = $this->getClientId();
        }
        return $frame;
    }

    /**
     * ActiveMq unsubscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param bool|false $durable
     * @return Frame
     */
    public function getUnsubscribeFrame($destination, $subscriptionId = null, $durable = false)
    {
        $frame = parent::getUnsubscribeFrame($destination, $subscriptionId);
        if ($durable) {
            $frame['activemq.subscriptionName'] = $this->getClientId();
        }
        return $frame;
    }

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

    /**
     * @inheritdoc
     */
    public function getNackFrame(Frame $frame, $transactionId = null, $requeue = null)
    {
        if ($requeue !== null) {
            throw new \LogicException(
                'requeue header not supported by ActiveMQ. Please read ActiveMQ DLQ documentation.'
            );
        }
        $nack = $this->createFrame('NACK');
        $nack['transaction'] = $transactionId;
        if ($this->hasVersion(Version::VERSION_1_2)) {
            $nack['id'] = $frame['ack'] ?: $frame->getMessageId();
        } else {
            $nack['message-id'] = $frame['ack'] ?: $frame->getMessageId();
            if ($this->hasVersion(Version::VERSION_1_1)) {
                $nack['subscription'] = $frame['subscription'];
            }
        }
        return $nack;
    }

    /**
     * Prefetch Size for subscriptions
     *
     * @return int
     */
    public function getPrefetchSize()
    {
        return $this->prefetchSize;
    }

    /**
     * Prefetch Size for subscriptions
     *
     * @param int $prefetchSize
     * @return ActiveMq
     */
    public function setPrefetchSize($prefetchSize)
    {
        $this->prefetchSize = $prefetchSize;
        return $this;
    }
}


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\ActiveMq\Mode;

use Stomp\Broker\ActiveMq\ActiveMq;
use Stomp\Broker\ActiveMq\Options;
use Stomp\Broker\Exception\UnsupportedBrokerException;
use Stomp\Client;

/**
 * ActiveMqMode
 *
 * @package Stomp\Broker\ActiveMq\Mode
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
abstract class ActiveMqMode
{
    /**
     * @var Client
     */
    protected $client;

    /**
     * @var Options
     */
    protected $options;

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

    /**
     * @return ActiveMq
     * @throws \Stomp\Broker\Exception\UnsupportedBrokerException
     */
    protected function getProtocol()
    {
        $protocol = $this->client->getProtocol();
        if (!$protocol instanceof ActiveMq) {
            throw new UnsupportedBrokerException($protocol, ActiveMq::class);
        }
        return $protocol;
    }

    /**
     * @return Options
     */
    public function getOptions()
    {
        return $this->options;
    }

    /**
     * @param Options $options
     * @return ActiveMqMode
     */
    public function setOptions(Options $options)
    {
        $this->options = $options;
        return $this;
    }
}


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\ActiveMq\Mode;

use Stomp\Client;
use Stomp\Exception\StompException;
use Stomp\States\Meta\Subscription;
use Stomp\Transport\Frame;

/**
 * DurableSubscription for ActiveMq.
 *
 * @package Stomp\Broker\ActiveMq\Mode
 * @author Jens Radtke <swefl.oss@fin-sn.de>l
 */
class DurableSubscription extends ActiveMqMode
{
    /**
     * @var Subscription
     */
    private $subscription;

    /**
     * Subscription state
     *
     * @var bool
     */
    private $active = false;

    /**
     * DurableSubscription constructor.
     * @param Client $client
     * @param string $topic
     * @param string $selector
     * @param string $ack
     * @throws StompException
     */
    public function __construct(Client $client, $topic, $selector = null, $ack = 'auto')
    {
        parent::__construct($client);
        if (!$client->getClientId()) {
            throw new StompException('Client must have been configured to use a specific clientId!');
        }
        $this->subscription = new Subscription($topic, $selector, $ack, $client->getClientId());
    }

    /**
     * Init the subscription.
     *
     * @return void
     */
    public function activate()
    {
        if (!$this->active) {
            $this->client->sendFrame(
                $this->getProtocol()->getSubscribeFrame(
                    $this->subscription->getDestination(),
                    $this->subscription->getSubscriptionId(),
                    $this->subscription->getAck(),
                    $this->subscription->getSelector(),
                    true
                )->addHeaders(
                    $this->options->getOptions()
                )
            );
            $this->active = true;
        }
    }

    /**
     * Mark durable subscription as offline.
     *
     * @see deactivate() if you want to indicate that the consumer is permanently removed.
     * @return void
     */
    public function inactive()
    {
        if ($this->active) {
            $this->client->sendFrame(
                $this->getProtocol()
                    ->getUnsubscribeFrame(
                        $this->subscription->getDestination(),
                        $this->subscription->getSubscriptionId()
                    )
            );
            $this->active = false;
        }
    }

    /**
     * Permanently remove durable subscription.
     *
     * @see inactive() if you just want to indicate that the consumer is offline now.
     * @return void
     */
    public function deactivate()
    {
        if ($this->active) {
            $this->inactive();
            $this->client->sendFrame(
                $this->getProtocol()
                    ->getUnsubscribeFrame(
                        $this->subscription->getDestination(),
                        $this->subscription->getSubscriptionId(),
                        true
                    )
            );
            $this->active = false;
        }
    }

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

    /**
     * Ack a frame.
     *
     * @param Frame $frame
     * @return void
     */
    public function ack(Frame $frame)
    {
        $this->client->sendFrame($this->getProtocol()->getAckFrame($frame));
    }

    /**
     * Nack a frame.
     *
     * @param Frame $frame
     * @return void
     */
    public function nack(Frame $frame)
    {
        $this->client->sendFrame($this->getProtocol()->getNackFrame($frame));
    }

    /**
     * Returns the Subscription details.
     *
     * @return Subscription
     */
    public function getSubscription()
    {
        return $this->subscription;
    }

    /**
     * Check if subscription is currently active.
     *
     * @return boolean
     */
    public function isActive()
    {
        return $this->active;
    }
}


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\ActiveMq;

use ArrayAccess;

/**
 * Options for ActiveMq Stomp
 *
 * For more details visit http://activemq.apache.org/stomp.html
 *
 * @package Stomp\Broker\ActiveMq
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class Options implements ArrayAccess
{
    private $extensions = [
        'activemq.dispatchAsync',
        'activemq.exclusive',
        'activemq.maximumPendingMessageLimit',
        'activemq.noLocal',
        'activemq.prefetchSize',
        'activemq.priority',
        'activemq.retroactive',
    ];

    private $options = [];

    /**
     * Options constructor.
     * @param array $options
     */
    public function __construct(array $options = [])
    {
        foreach ($options as $key => $value) {
            $this[$key] = $value;
        }
    }

    public function offsetExists($offset)
    {
        return isset($this->options[$offset]);
    }

    public function offsetGet($offset)
    {
        return $this->options[$offset];
    }

    public function offsetSet($offset, $value)
    {
        if (in_array($offset, $this->extensions, true)) {
            $this->options[$offset] = $value;
        }
    }

    public function offsetUnset($offset)
    {
        unset($this->options[$offset]);
    }

    public function getOptions()
    {
        return $this->options;
    }

    public function activateRetroactive()
    {
        $this['activemq.retroactive'] = 'true';
        return $this;
    }
    public function activateExclusive()
    {
        $this['activemq.exclusive'] = 'true';
        return $this;
    }

    public function activateDispatchAsync()
    {
        $this['activemq.dispatchAsync'] = 'true';
        return $this;
    }

    public function setPriority($priority)
    {
        $this['activemq.priority'] = $priority;
        return $this;
    }

    public function setPrefetchSize($size)
    {
        $this['activemq.prefetchSize'] = max($size, 1);
        return $this;
    }

    public function activateNoLocal()
    {
        $this['activemq.noLocal'] = 'true';
        return $this;
    }

    public function setMaximumPendingLimit($limit)
    {
        $this['activemq.maximumPendingMessageLimit'] = $limit;
        return $this;
    }
}


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\Apollo;

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

/**
 * Apollo Stomp dialect.
 *
 * @package Stomp
 * @author András Rutkai <riskawarrior@live.com>
 */
class Apollo extends Protocol
{
    /**
     * Apollo subscribe frame.
     *
     * @param string $destination
     * @param string $subscriptionId
     * @param string $ack
     * @param string $selector
     * @param boolean $durable durable subscription
     * @return \Stomp\Transport\Frame
     */
    public function getSubscribeFrame(
        $destination,
        $subscriptionId = null,
        $ack = 'auto',
        $selector = null,
        $durable = false
    ) {
        $frame = parent::getSubscribeFrame($destination, $subscriptionId, $ack, $selector);
        if ($this->hasClientId() && $durable) {
            $frame['persistent'] = 'true';
        }
        return $frame;
    }

    /**
     * Apollo 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;
    }

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

    /**
     * @inheritdoc
     */
    public function getNackFrame(Frame $frame, $transactionId = null, $requeue = null)
    {
        if ($requeue !== null) {
            throw new \LogicException('requeue header not supported');
        }
        $nack = $this->createFrame('NACK');
        $nack['transaction'] = $transactionId;
        if ($this->hasVersion(Version::VERSION_1_2)) {
            $nack['id'] = $frame['ack'] ?: $frame->getMessageId();
        } else {
            $nack['message-id'] = $frame['ack'] ?: $frame->getMessageId();
            if ($this->hasVersion(Version::VERSION_1_1)) {
                $nack['subscription'] = $frame['subscription'];
            }
        }
        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\Broker\Apollo\Mode;

use Stomp\Broker\Apollo\Apollo;
use Stomp\Broker\Exception\UnsupportedBrokerException;
use Stomp\Client;
use Stomp\States\Meta\Subscription;
use Stomp\Util\IdGenerator;

/**
 * QueueBrowser ApolloMq util to browse a queue without removing messages from it.
 *
 * @see http://activemq.apache.org/apollo/documentation/stomp-manual.html#Browsing_Subscriptions
 *
 * @package Stomp\Broker\Apollo\Mode
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class QueueBrowser
{
    /**
     * @var Client
     */
    private $client;

    /**
     * @var bool
     */
    private $stopOnEnd;

    /**
     * @var Subscription
     */
    private $subscription;

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

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

    /**
     * QueueBrowser constructor.
     *
     * @param Client $client
     * @param string $destination
     * @param bool $stopOnEnd
     */
    public function __construct(Client $client, $destination, $stopOnEnd = true)
    {
        $this->stopOnEnd = $stopOnEnd;
        $this->client = $client;
        $this->stopOnEnd = $stopOnEnd;
        $this->subscription = new Subscription($destination, null, 'auto', IdGenerator::generateId());
    }

    /**
     * Protocol
     * @return Apollo
     * @throws UnsupportedBrokerException
     */
    final protected function getProtocol()
    {
        $protocol = $this->client->getProtocol();
        if (!$protocol instanceof Apollo) {
            throw new UnsupportedBrokerException($protocol, Apollo::class);
        }
        return $protocol;
    }

    /**
     * Headers used in subscribe.
     *
     * @return array
     */
    protected function getHeader()
    {
        return [
            'browser' => 'true',
            'browser-end' => $this->stopOnEnd ? 'true' : 'false',
        ];
    }

    /**
     * Initialize subscription.
     *
     * @return void
     */
    public function subscribe()
    {
        if (!$this->active) {
            $this->reachedEnd = false;
            $this->client->sendFrame(
                $this->getProtocol()->getSubscribeFrame(
                    $this->subscription->getDestination(),
                    $this->subscription->getSubscriptionId(),
                    $this->subscription->getAck(),
                    $this->subscription->getSelector(),
                    false
                )->addHeaders($this->getHeader())
            );
            $this->active = true;
        }
    }

    /**
     * End subscription.
     *
     * @return void
     */
    public function unsubscribe()
    {
        if ($this->active) {
            $this->client->sendFrame(
                $this->getProtocol()->getUnsubscribeFrame(
                    $this->subscription->getDestination(),
                    $this->subscription->getSubscriptionId(),
                    false
                )
            );
            $this->active = false;
        }
    }

    /**
     * Read next message.
     *
     * @return bool|false|\Stomp\Transport\Frame
     */
    public function read()
    {
        if (!$this->active || $this->reachedEnd) {
            return false;
        }
        if ($frame = $this->client->readFrame()) {
            if ($this->stopOnEnd && $frame['browser'] == 'end') {
                $this->reachedEnd = true;
                return false;
            }
        }
        return $frame;
    }

    /**
     * Last message was received (can only be true if 'StopAtEnd' is enabled!)
     *
     * @return boolean
     */
    public function hasReachedEnd()
    {
        return $this->reachedEnd;
    }

    /**
     * Subscription has been initialized.
     *
     * @return boolean
     */
    public function isActive()
    {
        return $this->active;
    }

    /**
     * Subscription details.
     *
     * @return Subscription
     */
    public function getSubscription()
    {
        return $this->subscription;
    }

    /**
     * @inheritdoc
     */
    public function __destruct()
    {
        IdGenerator::releaseId($this->subscription->getSubscriptionId());
    }
}


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\Apollo\Mode;

use Stomp\Client;

/**
 * SequenceQueueBrowser ApolloMq util to browse a queue sequence based without removing messages from it.
 *
 * @see http://activemq.apache.org/apollo/documentation/stomp-manual.html
 *      #Using_Queue_Browsers_to_Implement_Durable_Topic_Subscriptions
 *
 * @package Stomp\Broker\Apollo\Mode
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class SequenceQueueBrowser extends QueueBrowser
{
    /**
     * Browser start at head at current queue
     */
    const START_HEAD = 0;
    /**
     * Browser start at for new messages in queue
     */
    const START_NEW = -1;

    /**
     * @var int
     */
    private $startAt;

    /**
     * Current offset.
     *
     * @var null|int
     */
    private $seq;

    /**
     * SequenceQueueBrowser constructor.
     *
     * @param Client $client
     * @param string $destination
     * @param int $startAt
     * @param bool $stopOnEnd
     */
    public function __construct(Client $client, $destination, $startAt = self::START_HEAD, $stopOnEnd = true)
    {
        $this->startAt = $startAt;
        parent::__construct($client, $destination, $stopOnEnd);
    }

    /**
     * @inheritdoc
     */
    protected function getHeader()
    {
        return parent::getHeader() + ['include-seq' => 'seq', 'from-seq' => $this->startAt];
    }

    /**
     * @inheritdoc
     */
    public function read()
    {
        if ($frame = parent::read()) {
            $this->seq = $frame['seq'];
        }
        return $frame;
    }

    /**
     * Returns the last received sequence.
     *
     * @return null|int
     */
    public function getSeq()
    {
        return $this->seq;
    }
}


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\Exception;

use Stomp\Exception\StompException;
use Stomp\Protocol\Protocol;

/**
 * UnsupportedBrokerException
 *
 * Indicate that given broker is not supported.
 *
 * @package Stomp\Broker\Exception
 * @author Jens Radtke <swefl.oss@fin-sn.de>
 */
class UnsupportedBrokerException extends StompException
{

    /**
     * @param Protocol $detectedProtocol
     * @param string $expectedProtocol
     */
    public function __construct(Protocol $detectedProtocol, $expectedProtocol)
    {
        parent::__construct(
            sprintf('The current broker (%s) is no %s.', get_class($detectedProtocol), $expectedProtocol)
        );
    }
}


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\OpenMq;

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

/**
 * OpenMq Stomp dialect.
 *
 * @package Stomp
 * @author Markus Staab <maggus.staab@googlemail.com>
 */
class OpenMq extends Protocol
{
    /**
     * @inheritdoc
     */
    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();
        }
        // spec quote: "ACK should always specify a "subscription" header for the subscription id
        //              that the message to be acked was delivered to ."
        // see https://mq.java.net/4.4-content/stomp-funcspec.html
        $ack['subscription'] = $frame['subscription'];
        return $ack;
    }
}


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;
    }
}


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);
    }
}