code:
<?php
declare(strict_types=1);

namespace Interop\Queue;

interface Processor
{
    /**
     * Use this constant when the message is processed successfully and the message could be removed from the queue.
     */
    const ACK = 'enqueue.ack';

    /**
     * Use this constant when the message is not valid or could not be processed
     * The message is removed from the queue.
     */
    const REJECT = 'enqueue.reject';

    /**
     * Use this constant when the message is not valid or could not be processed right now but we can try again later
     * The original message is removed from the queue but a copy is published to the queue again.
     */
    const REQUEUE = 'enqueue.requeue';

    /**
     * The method has to return either self::ACK, self::REJECT, self::REQUEUE string.
     *
     * The method also can return an object.
     * It must implement __toString method and the method must return one of the constants from above.
     *
     * @param Message $message
     * @param Context $context
     *
     * @return string|object with __toString method implemented
     */
    public function process(Message $message, Context $context);
}


code:
<?php
declare(strict_types=1);

namespace Interop\Queue;

use Interop\Queue\Exception\DeliveryDelayNotSupportedException;
use Interop\Queue\Exception\InvalidDestinationException;
use Interop\Queue\Exception\InvalidMessageException;
use Interop\Queue\Exception\PriorityNotSupportedException;
use Interop\Queue\Exception\TimeToLiveNotSupportedException;

interface Producer
{
    /**
     * @throws Exception                   if the provider fails to send the message due to some internal error
     * @throws InvalidDestinationException if a client uses this method with an invalid destination
     * @throws InvalidMessageException     if an invalid message is specified
     */
    public function send(Destination $destination, Message $message): void;

    /**
     * Sets the minimum length of time in milliseconds that must elapse after a message is sent before the provider may deliver the message to a consumer.
     * deliveryDelay is set to null default.
     *
     * The delivery delay is in milliseconds. Use null to unset delivery delay and use transport's mode
     *
     * @throws DeliveryDelayNotSupportedException if producer does not support delivery delay feature
     */
    public function setDeliveryDelay(int $deliveryDelay = null): self;

    /**
     * Gets the minimum length of time in milliseconds that must elapse after a message is sent before the provider may deliver the message to a consumer.
     *
     * @return int|null the delivery delay in milliseconds.
     */
    public function getDeliveryDelay(): ?int;

    /**
     * Specifies the priority of messages that are sent using this Producer
     * The API defines ten levels of priority value, with 0 as the lowest priority and 9 as the highest.
     * Clients should consider priorities 0-4 as gradations of normal priority and priorities 5-9 as gradations of expedited priority.
     * Priority is set to null by default.
     *
     * Use null to unset priority and use transport's mode
     *
     * @throws PriorityNotSupportedException if producer does not support priority feature
     */
    public function setPriority(int $priority = null): self;

    /**
     * Return the priority of messages that are sent using this Producer
     */
    public function getPriority(): ?int;

    /**
     * Specifies the time to live of messages that are sent using this Producer.
     * This is used to determine the expiration time of a message.
     * The expiration time of a message is the sum of the message's time to live and the time it is sent.
     * Clients should not receive messages that have expired; however, This interop does not guarantee that this will not happen.
     * A provider should do its best to accurately expire messages; however, it does not define the accuracy provided.
     * It is not acceptable to simply ignore time-to-live.
     * timeToLive is set to null default.
     *
     * The message time to live to be used, in milliseconds; a value of zero means that a message never expires. Use null to unset time to live and use transport's mode
     *
     * @throws TimeToLiveNotSupportedException if producer does not support time to live feature
     */
    public function setTimeToLive(int $timeToLive = null): self;

    /**
     * Returns the time to live of messages that are sent using this JMSProducer.
     *
     * @return int|null the message time to live in milliseconds; a value of zero means that a message never expires.
     */
    public function getTimeToLive(): ?int;
}

repo_name: spookie/gnu

license: mit


code:
<?php
declare(strict_types=1);

namespace Interop\Queue;

interface Queue extends Destination
{
    /**
     * Gets the name of this queue. This is a destination one consumes messages from.
     */
    public function getQueueName(): string;
}


code:
<?php
declare(strict_types=1);

namespace Interop\Queue;

interface SubscriptionConsumer
{
    /**
     * The timeout is in milliseconds.
     * Set zero to consume endlessly or till a consumer returns false.
     */
    public function consume(int $timeout = 0): void;

    /**
     * Notify broker that the client is interested in consuming messages from this queue.
     *
     * A callback function to which the consumed message will be passed.
     * The function must accept at a minimum one parameter, an \Interop\Queue\Message object,
     * and an optional second parameter the \Interop\Queue\Consumer from which the message was
     * consumed. The consumer will not return the processing thread back to
     * the PHP script until the callback function returns FALSE.
     */
    public function subscribe(Consumer $consumer, callable $callback): void;

    public function unsubscribe(Consumer $consumer): void;

    public function unsubscribeAll(): void;
}


code:
<?php
declare(strict_types=1);

namespace Interop\Queue;

/**
 * A Topic object encapsulates a provider-specific topic name.
 * It is the way a client specifies the identity of a topic to transport methods.
 * For those methods that use a Destination as a parameter, a Topic object may used as an argument.
 *
 * @see https://docs.oracle.com/javaee/7/api/javax/jms/Topic.html
 */
interface Topic extends Destination
{
    /**
     * Gets the name of this topic. This is a destination one sends messages to.
     */
    public function getTopicName(): string;
}


code:
<?php
namespace Interop\Queue\Tests\Impl;

use Interop\Queue\Impl\ConsumerPollingTrait;
use Interop\Queue\Message;
use PHPUnit\Framework\TestCase;

class ConsumerPollingTest extends TestCase
{

    public function getConsumer(callable $receiveNoWait)
    {
        return new class($receiveNoWait)
        {
            use ConsumerPollingTrait;

            private $count = 0;
            private $callable;

            public function __construct(callable $receiveNoWait)
            {
                $this->callable = $receiveNoWait;
            }

            public function receiveNoWait(): ?Message
            {
                $callable = $this->callable;
                return $callable($this->count);
            }
        };
    }

    public function testShouldTimeout()
    {
        $consumer = $this->getConsumer(function() {
            return null;
        });

        $start = microtime(true);

        $consumer->receive(250);

        $timer = microtime(true) - $start;

        $this->assertGreaterThan(0.250, $timer);
        $this->assertLessThan(0.255, $timer);
    }

    public function testShouldCallThreeTimes()
    {
        $message = $this->getMockForAbstractClass(Message::class);

        $consumer = $this->getConsumer(function(int &$count) use($message) {
            $count++;
            if(3 === $count) {
                return $message;
            }
            return null;
        });
        $consumer->setPollingInterval(100);

        $start = microtime(true);

        $this->assertSame($message, $consumer->receive(250));

        $timer = microtime(true) - $start;

        $this->assertGreaterThan(0.200, $timer);
        $this->assertLessThan(0.205, $timer);
    }
}


code:
<?php
declare(strict_types=1);

namespace Interop\Queue\Tests\Impl;

use Interop\Queue\Impl\ConsumerVisibilityTimeoutTrait;
use PHPUnit\Framework\TestCase;

class ConsumerVisibilityTimeoutTest extends TestCase
{

    public function testShoudGetAndSetVisibilityTimeout()
    {
        $consumer = new class {
            use ConsumerVisibilityTimeoutTrait;
        };

        $this->assertNull($consumer->getVisibilityTimeout());

        $consumer->setVisibilityTimeout(123);

        $this->assertSame(123, $consumer->getVisibilityTimeout());
    }

}


code:
<?php
declare(strict_types=1);

namespace Interop\Queue\Tests\Impl;

use Interop\Queue\Message;
use Interop\Queue\Spec\MessageSpec;
use Interop\Queue\Impl\MessageTrait;

class MessageTest extends MessageSpec
{
    protected function createMessage(): Message
    {
        return new class implements Message
        {
            use MessageTrait;

            public function __construct(string $body = '', array $properties = [], array $headers = [])
            {
                $this->body = $body;
                $this->properties = $properties;
                $this->headers = $headers;

                $this->redelivered = false;
            }
        };
    }
}


code:
getallheaders
=============

PHP `getallheaders()` polyfill. Compatible with PHP >= 5.3.

[![Build Status](https://travis-ci.org/ralouphie/getallheaders.svg?branch=master)](https://travis-ci.org/ralouphie/getallheaders)
[![Coverage Status](https://coveralls.io/repos/ralouphie/getallheaders/badge.png?branch=master)](https://coveralls.io/r/ralouphie/getallheaders?branch=master)
[![Latest Stable Version](https://poser.pugx.org/ralouphie/getallheaders/v/stable.png)](https://packagist.org/packages/ralouphie/getallheaders)
[![Latest Unstable Version](https://poser.pugx.org/ralouphie/getallheaders/v/unstable.png)](https://packagist.org/packages/ralouphie/getallheaders)
[![License](https://poser.pugx.org/ralouphie/getallheaders/license.png)](https://packagist.org/packages/ralouphie/getallheaders)

This is a simple polyfill for [`getallheaders()`](http://www.php.net/manual/en/function.getallheaders.php).

## Install

For PHP version **`>= 5.6`**:

```
composer require ralouphie/getallheaders
```

For PHP version **`< 5.6`**:

```
composer require ralouphie/getallheaders "^2"
```


ralouphie/getallheaders

description: A polyfill for getallheaders.

license: MIT

getallheaders\Tests\: tests/


code:
<?php

if (!function_exists('getallheaders')) {

    /**
     * Get all HTTP header key/values as an associative array for the current request.
     *
     * @return string[string] The HTTP header key/value pairs.
     */
    function getallheaders()
    {
        $headers = array();

        $copy_server = array(
            'CONTENT_TYPE'   => 'Content-Type',
            'CONTENT_LENGTH' => 'Content-Length',
            'CONTENT_MD5'    => 'Content-Md5',
        );

        foreach ($_SERVER as $key => $value) {
            if (substr($key, 0, 5) === 'HTTP_') {
                $key = substr($key, 5);
                if (!isset($copy_server[$key]) || !isset($_SERVER[$key])) {
                    $key = str_replace(' ', '-', ucwords(strtolower(str_replace('_', ' ', $key))));
                    $headers[$key] = $value;
                }
            } elseif (isset($copy_server[$key])) {
                $headers[$copy_server[$key]] = $value;
            }
        }

        if (!isset($headers['Authorization'])) {
            if (isset($_SERVER['REDIRECT_HTTP_AUTHORIZATION'])) {
                $headers['Authorization'] = $_SERVER['REDIRECT_HTTP_AUTHORIZATION'];
            } elseif (isset($_SERVER['PHP_AUTH_USER'])) {
                $basic_pass = isset($_SERVER['PHP_AUTH_PW']) ? $_SERVER['PHP_AUTH_PW'] : '';
                $headers['Authorization'] = 'Basic ' . base64_encode($_SERVER['PHP_AUTH_USER'] . ':' . $basic_pass);
            } elseif (isset($_SERVER['PHP_AUTH_DIGEST'])) {
                $headers['Authorization'] = $_SERVER['PHP_AUTH_DIGEST'];
            }
        }

        return $headers;
    }

}


code:
PHP RabbitMQ Management Api
===========================

A simple object oriented wrapper for the [RabbitMQ Management HTTP Api](http://hg.rabbitmq.com/rabbitmq-management/raw-file/rabbitmq_v3_0_3/priv/www/api/index.html) in PHP 5.3

Uses [PHP-HTTP](http://docs.php-http.org/en/latest/index.html) for requests.

Installation
------------

Installable through composer via:

```bash
$ composer require richardfullmer/rabbitmq-management-api
```

Additionally, you require a [httplug compatible client](http://docs.php-http.org/en/latest/clients.html).

For example, use the guzzle6 adapter:

```bash
$ composer require php-http/guzzle6-adapter
```

Basic Usage
-----------

```php
<?php

use RabbitMq\ManagementApi\Client;

require_once __DIR__ . '/../vendor/autoload.php';

$client = new Client();
$queue = $client->queues()->get('/', 'sample-messages-queue');
$response = $client->exchanges()->publish('/', 'sample-messages', array(
    'properties' => array(),
    'routing_key' => '',
    'payload' => 'This is a test',
    'payload_encoding' => 'string'
));

if ($response['routed']) {
    print 'Message delivered';
}
```

License
-------

php-rabbitmq-management-api is licensed under the MIT License - see the LICENSE file for details

Credits
-------

Structure from [KnpLabs php-github-api](https://github.com/KnpLabs/php-github-api)

Rabbit's Excellent Message Queue


richardfullmer/rabbitmq-management-api

type: library

description: An object oriented wrapper for the RabbitMQ Management HTTP Api

keywords:
rest
message queue
rabbitmq

RabbitMq\ManagementApi\: src/


code:
<?php

namespace RabbitMq\ManagementApi\Api;

use RabbitMq\ManagementApi\Client;

/**
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class AbstractApi
{
    protected $client;

    public function __construct(Client $client)
    {
        $this->client = $client;
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Binding extends AbstractApi
{
    /**
     * A list of all bindings.
     *
     * OR
     *
     * A list of all bindings in a given virtual host.
     *
     * @param string|null $vhost
     * @return array
     */
    public function all($vhost = null)
    {
        if ($vhost) {
            return $this->client->send(sprintf('/api/bindings/%s', urlencode($vhost)));
        } else {
            return $this->client->send('/api/bindings');
        }
    }

    /**
     * A list of all bindings between an exchange and a queue. Remember, an exchange and a queue can be bound together
     * many times!
     *
     * @param string $vhost
     * @param string $exchange
     * @param string $queue
     * @return array
     */
    public function binding($vhost, $exchange, $queue)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/q/%s', urlencode($vhost), urlencode($exchange), urlencode($queue)));
    }

    /**
     * A list of all bindings between two exchanges. Remember, two exchanges can be bound together many times with
     * different parameters!
     *
     * @param string $vhost
     * @param string $source
     * @param string $destination
     *
     * @return array
     */
    public function exchangeBinding($vhost, $source, $destination)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/e/%s', urlencode($vhost), urlencode($source), urlencode($destination)));
    }

    /**
     *  To create a new binding, POST to this URI. You will need a body looking something like this:
     *
     * {
     *     "routing_key": "my_routing_key",
     *     "arguments": []
     * }
     *
     * All keys are optional. The response will contain a Location header telling you the URI of your new binding.
     *
     * @param string $vhost
     * @param string $exchange
     * @param string $queue
     * @param string|null $routingKey
     * @param array|null $arguments
     * @return array
     */
    public function create($vhost, $exchange, $queue, $routingKey = null, array $arguments = null)
    {
        $parameters = array();

        if ($routingKey) {
            $parameters['routing_key'] = $routingKey;
        } else {
            $parameters['routing_key'] = '';
        }
        if ($arguments) {
            $parameters['arguments'] = $arguments;
        }

        return $this->client->send(sprintf('/api/bindings/%s/e/%s/q/%s', urlencode($vhost), urlencode($exchange), urlencode($queue)), 'POST', [], $parameters);
    }

    /**
     *  To create a new exchange to exchange binding, POST to this URI. You will need a body looking something like this:
     *
     * {
     *     "routing_key": "my_routing_key",
     *     "arguments": []
     * }
     *
     * All keys are optional. The response will contain a Location header telling you the URI of your new binding.
     *
     * @param string $vhost
     * @param string $source
     * @param string $destination
     * @param string|null $routingKey
     * @param array|null $arguments
     * @return array
     */
    public function createExchange($vhost, $source, $destination, $routingKey = null, array $arguments = null)
    {
        $parameters = array();

        if ($routingKey) {
            $parameters['routing_key'] = $routingKey;
        } else {
            $parameters['routing_key'] = '';
        }
        if ($arguments) {
            $parameters['arguments'] = $arguments;
        }

        return $this->client->send(sprintf('/api/bindings/%s/e/%s/e/%s', urlencode($vhost), urlencode($source), urlencode($destination)), 'POST', [], $parameters);
    }

    /**
     * An individual binding between an exchange and a queue. The props part of the URI is a "name" for the binding
     * composed of its routing key and a hash of its arguments.
     *
     * @param string $vhost
     * @param string $exchange
     * @param string $queue
     * @param string $props
     * @return array
     */
    public function get($vhost, $exchange, $queue, $props)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/q/%s/%s', urlencode($vhost), urlencode($exchange), urlencode($queue), urlencode($props)));
    }

    /**
     * An individual binding between two exchanges. The props part of the URI is a "name" for the binding
     * composed of its routing key and a hash of its arguments.
     *
     * @param string $vhost
     * @param string $source
     * @param string $destination
     * @param string $props
     * @return array
     */
    public function getExchange($vhost, $source, $destination, $props)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/e/%s/%s', urlencode($vhost), urlencode($source), urlencode($destination), urlencode($props)));
    }

    /**
     * Remove an individual binding between an exchange and a queue.
     *
     * @param string $vhost
     * @param string $exchange
     * @param string $queue
     * @param string $props
     * @return array
     */
    public function delete($vhost, $exchange, $queue, $props)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/q/%s/%s', urlencode($vhost), urlencode($exchange), urlencode($queue), urlencode($props)), 'DELETE');
    }

    /**
     * Remove an individual binding between two exchanges.
     *
     * @param string $vhost
     * @param string $source
     * @param string $destination
     * @param string $props
     * @return array
     */
    public function deleteExchange($vhost, $source, $destination, $props)
    {
        return $this->client->send(sprintf('/api/bindings/%s/e/%s/e/%s/%s', urlencode($vhost), urlencode($source), urlencode($destination), urlencode($props)), 'DELETE');
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Channel
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Channel extends AbstractApi
{
    /**
     * A list of all open channels.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/channels');
    }

    /**
     * Details about an individual channel.
     *
     * @param string $channel
     * @return array
     */
    public function get($channel)
    {
        return $this->client->send(sprintf('/api/channels/%s', urlencode($channel)));
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Connection
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Connection extends AbstractApi
{
    /**
     * A list of all open connections.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/connections');
    }

    /**
     * An individual connection.
     *
     * @param string $name
     * @return array|int
     */
    public function get($name)
    {
        return $this->client->send(sprintf('/api/connections/%s', urlencode($name)));
    }

    /**
     *  Deleting a connection will close it.
     *
     * @param string $name
     * @return array|int
     */
    public function delete($name)
    {
        return $this->client->send(sprintf('/api/connections/%s', urlencode($name)), 'DELETE');
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

use RabbitMq\ManagementApi\Exception\InvalidArgumentException;

/**
 * Exchange
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Exchange extends AbstractApi
{
    /**
     * A list of all exchanges.
     *
     * OR
     *
     * A list of all exchanges in a given virtual host.
     *
     * @param null|string $vhost
     * @return array
     */
    public function all($vhost = null)
    {
        if ($vhost) {
            return $this->client->send(sprintf('/api/exchanges/%s', urlencode($vhost)));
        } else {
            return $this->client->send('/api/exchanges');
        }
    }

    /**
     * An individual exchange.
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function get($vhost, $name)
    {
        return $this->client->send(sprintf('/api/exchanges/%s/%s', urlencode($vhost), urlencode($name)));
    }

    /**
     * To create an exchange, you will need a body looking something like this:
     *
     * {
     *     "type": "direct",
     *     "auto_delete": false,
     *     "durable": true,
     *     "internal": false,
     *     "arguments": []
     * }
     *
     * The 'type' key is mandatory; other keys are optional.
     *
     * @param string $vhost
     * @param string $name
     * @param array  $exchange
     * @return array
     * @throws \RabbitMq\ManagementApi\Exception\InvalidArgumentException
     */
    public function create($vhost, $name, array $exchange)
    {
        if (!isset($exchange['type'])) {
            throw new InvalidArgumentException("Error creating exchange: Exchange key 'type' is mandatory");
        }

        return $this->client->send(sprintf('/api/exchanges/%s/%s', urlencode($vhost), urlencode($name)), 'PUT', [], $exchange);
    }

    /**
     * Delete an exchange
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function delete($vhost, $name)
    {
        return $this->client->send(sprintf('/api/exchanges/%s/%s', urlencode($vhost), urlencode($name)), 'DELETE');
    }

    /**
     * A list of all bindings in which a given exchange is the source.
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function sourceBindings($vhost, $name)
    {
        return $this->client->send(sprintf('/api/exchanges/%s/%s/bindings/source', urlencode($vhost), urlencode($name)));
    }

    /**
     * A list of all bindings in which a given exchange is the destination.
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function destinationBindings($vhost, $name)
    {
        return $this->client->send(sprintf('/api/exchanges/%s/%s/bindings/destination', urlencode($vhost), urlencode($name)));
    }

    /**
     * Publish a message to a given exchange. You will need a body looking something like:
     *
     * {
     *     "properties": {},
     *     "routing_key": "my key",
     *     "payload": "my body",
     *     "payload_encoding":"string"
     * }
     *
     * All keys are mandatory. The payload_encoding key should be either "string" (in which case the payload will be
     * taken to be the UTF-8 encoding of the payload field) or "base64" (in which case the payload field is taken to be
     * base64 encoded).
     *
     * If the message is published successfully, the response will look like:
     *
     * {"routed": true}
     *
     * routed will be true if the message was sent to at least one queue.
     *
     * Please note that the publish / get paths in the HTTP API are intended for injecting test messages, diagnostics etc - they do not implement reliable delivery and so should be treated as a sysadmin's tool rather than a general API for messaging.
     *
     * @param string $vhost
     * @param string $name
     * @param array $message
     * @throws InvalidArgumentException
     * @return array
     */
    public function publish($vhost, $name, array $message)
    {
        if (!isset($message['properties'])) {
            throw new InvalidArgumentException("Error publishing to exchange: Message key 'properties' is mandatory");
        } elseif (!isset($message['routing_key'])) {
            throw new InvalidArgumentException("Error publishing to exchange: Message key 'routing_key' is mandatory");
        } elseif (!isset($message['payload'])) {
            throw new InvalidArgumentException("Error publishing to exchange: Message key 'payload' is mandatory");
        } elseif (!isset($message['payload_encoding'])) {
            throw new InvalidArgumentException("Error publishing to exchange: Message key 'payload_encoding' is mandatory");
        }

        return $this->client->send(sprintf('/api/exchanges/%s/%s/publish', urlencode($vhost), urlencode($name)), 'POST', [], $message);
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Nodes
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Node extends AbstractApi
{
    /**
     * A list of nodes in the RabbitMQ cluster.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/nodes');
    }

    /**
     * An individual node in the RabbitMQ cluster. Add "?memory=true" to get memory statistics.
     *
     * @param string $name
     * @param bool   $memory
     * @return array
     */
    public function get($name, $memory = false)
    {
        return $this->client->send(sprintf('/api/nodes/%s%s', urlencode($name), $memory ? '?memory=true' : ''));
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Parameter
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Parameter extends AbstractApi
{
    /**
     * A list of all parameters.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/parameters');
    }

    /**
     * A list of all parameters for a given component and virtual host.
     *
     * @param string $component
     * @param string|null $vhost
     * @param string|null $name
     * @return array
     */
    public function get($component, $vhost = null, $name = null)
    {
        if ($vhost && $name) {
            return $this->client->send(sprintf('/api/parameters/%s/%s/%s', urlencode($component), urlencode($vhost), urlencode($name)));
        } elseif ($vhost) {
            return $this->client->send(sprintf('/api/parameters/%s/%s', urlencode($component), urlencode($vhost)));
        } else {
            return $this->client->send(sprintf('/api/parameters/%s', urlencode($component)));
        }
    }

    /**
     * To PUT a parameter, you will need a body looking something like this:
     *
     * {
     *     "vhost": "/",
     *     "component": "federation",
     *     "name": "local_username",
     *     "value": "guest"
     * }
     *
     * @param string $component
     * @param string $vhost
     * @param string $name
     * @param array $parameter
     * @return array
     */
    public function create($component, $vhost, $name, array $parameter)
    {
        return $this->client->send(sprintf('/api/parameters/%s/%s/%s', urlencode($component), urlencode($vhost), urlencode($name)), 'PUT', [], $parameter);
    }

    /**
     * Delete a parameter
     *
     * @param string $component
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function delete($component, $vhost, $name)
    {
        return $this->client->send(sprintf('/api/parameters/%s/%s/%s', urlencode($component), urlencode($vhost), urlencode($name)), 'DELETE');
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

use RabbitMq\ManagementApi\Exception\InvalidArgumentException;

/**
 * Permission
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Permission extends AbstractApi
{
    /**
     * A list of all permissions for all users.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/permissions');
    }

    /**
     * An individual permission of a user and virtual host.
     *
     * @param string $vhost
     * @param string $user
     * @return array
     */
    public function get($vhost, $user)
    {
        return $this->client->send(sprintf('/api/permissions/%s/%s', urlencode($vhost), urlencode($user)));
    }

    /**
     * To PUT a permission, you will need a body looking something like this:
     *
     * {"configure":".*","write":".*","read":".*"}
     *
     * All keys are mandatory.
     *
     * @param string $vhost
     * @param string $user
     * @param array $permission
     * @return array
     * @throws InvalidArgumentException
     */
    public function create($vhost, $user, array $permission)
    {
        if (!isset($permission['configure']) || !isset($permission['write']) || !isset($permission['read'])) {
            throw new InvalidArgumentException("Error creating permission: 'configure', 'write', and 'read' permissions must be properly set.");
        }

        return $this->client->send(sprintf('/api/permissions/%s/%s', urlencode($vhost), urlencode($user)), 'PUT', [], $permission);
    }

    /**
     * Delete a specific set of permissions
     *
     * @param string $vhost
     * @param string $user
     * @return array
     */
    public function delete($vhost, $user)
    {
        return $this->client->send(sprintf('/api/permissions/%s/%s', urlencode($vhost), urlencode($user)), 'DELETE');
    }

}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Policy
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Policy extends AbstractApi
{
    /**
     * A list of all policies.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/policies');
    }

    /**
     * A list of all policies in a given virtual host.
     *
     * OR
     *
     * An individual policy.
     *
     * @param string $vhost
     * @param string|null $name
     * @return array
     */
    public function get($vhost, $name = null)
    {
        if ($name) {
            return $this->client->send(sprintf('/api/policies/%s/%s', urlencode($vhost), urlencode($name)));
        }

        return $this->client->send(sprintf('/api/policies/%s', urlencode($vhost)));
    }

    /**
     * To PUT a policy, you will need a body looking something like this:
     *
     * {"pattern":"^amq.", "definition": {"federation-upstream-set":"all"}, "priority":0}
     *
     * @param string $vhost
     * @param string $name
     * @param array $policy
     * @return array
     */
    public function create($vhost, $name, array $policy)
    {
        return $this->client->send(sprintf('/api/policies/%s/%s', urlencode($vhost), urlencode($name)), 'PUT', [], $policy);
    }

    /**
     * Delete a policy
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function delete($vhost, $name)
    {
        return $this->client->send(sprintf('/api/policies/%s/%s', urlencode($vhost), urlencode($name)), 'DELETE');
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Queue
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Queue extends AbstractApi
{
    /**
     * A list of all queues.
     *
     * OR
     *
     * A list of all queues in a given virtual host.
     *
     * @param string|null $vhost
     * @return array
     */
    public function all($vhost = null)
    {
        if ($vhost) {
            return $this->client->send(sprintf('/api/queues/%s', urlencode($vhost)));
        } else {
            return $this->client->send('/api/queues');
        }
    }

    /**
     * An individual queue.
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function get($vhost, $name)
    {
        return $this->client->send(sprintf('/api/queues/%s/%s', urlencode($vhost), urlencode($name)));
    }

    /**
     * To PUT a queue, you will need a body looking something like this:
     *
     * {
     *     "auto_delete": false,
     *     "durable": true,
     *     "arguments":[],
     *     "node":"rabbit@smacmullen"
     * }
     *
     * All keys are optional.
     *
     * @param string $vhost
     * @param string $name
     * @param array $queue
     * @return array
     */
    public function create($vhost, $name, array $queue)
    {
        return $this->client->send(sprintf('/api/queues/%s/%s', urlencode($vhost), urlencode($name)), 'PUT', [], $queue);
    }

    /**
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function delete($vhost, $name)
    {
        return $this->client->send(sprintf('/api/queues/%s/%s', urlencode($vhost), urlencode($name)), 'DELETE');
    }

    /**
     * A list of all bindings on a given queue.
     *
     * @param string $vhost
     * @param string $queue
     * @return array
     */
    public function bindings($vhost, $queue)
    {
        return $this->client->send(sprintf('/api/queues/%s/%s/bindings', urlencode($vhost), urlencode($queue)));
    }

    /**
     * Contents of a queue. DELETE to purge. Note you can't GET this.
     *
     * @param string $vhost
     * @param string $name
     * @return array
     */
    public function purgeMessages($vhost, $name)
    {
        return $this->client->send(sprintf('/api/queues/%s/%s/contents', urlencode($vhost), urlencode($name)), 'DELETE');
    }

    /**
     * Get messages from a queue. (This is not an HTTP GET as it will alter the state of the queue.) You should post a
     * body looking like:
     *
     * {"count":5,"requeue":true,"encoding":"auto","truncate":50000}
     *
     * count controls the number of messages to get. You may get fewer messages than this if the queue cannot
     * immediately provide them.
     *
     * requeue determines whether the messages will be removed from the queue. If requeue is true they will be
     * requeued - but their position in the queue may change and their redelivered flag will be set.
     *
     * encoding must be either "auto" (in which case the payload will be returned as a string if it is valid UTF-8, and
     * base64 encoded otherwise), or "base64" (in which case the payload will always be base64 encoded).
     *
     * If truncate is present it will truncate the message payload if it is larger than the size given (in bytes).
     *
     * truncate is optional; all other keys are mandatory.
     *
     * Please note that the publish / get paths in the HTTP API are intended for injecting test messages, diagnostics
     * etc - they do not implement reliable delivery and so should be treated as a sysadmin's tool rather than a general
     * API for messaging.
     *
     * @param string $vhost
     * @param string $name
     * @param integer $count
     * @param bool $requeue
     * @param string $encoding
     * @param null|integer $truncate
     * @return array
     */
    public function retrieveMessages($vhost, $name, $count = 5, $requeue = true, $encoding = 'auto', $truncate = null)
    {
        $parameters = array(
            'count' => $count,
            'requeue' => $requeue,
            'encoding' => $encoding
        );

        if ($truncate) {
            $parameters['truncate'] = $truncate;
        }

        return $this->client->send(sprintf('/api/queues/%s/%s/get', urlencode($vhost), urlencode($name)), 'POST', [], $parameters);
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * User
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class User extends AbstractApi
{
    /**
     * A list of all users.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/users');
    }

    /**
     * An individual user.
     *
     * @param string $name
     * @return array
     */
    public function get($name)
    {
        return $this->client->send(sprintf('/api/users/%s', urlencode($name)));
    }

    /**
     *  To PUT a user, you will need a body looking something like this:
     *
     * {"password":"secret","tags":"administrator"}
     *
     * or:
     *
     * {"password_hash":"2lmoth8l4H0DViLaK9Fxi6l9ds8=", "tags":"administrator"}
     *
     * The tags key is mandatory. Either password or password_hash must be set. Setting password_hash to "" will
     * ensure the user cannot use a password to log in. tags is a comma-separated list of tags for the user. Currently
     * recognised tags are "administrator", "monitoring" and "management".
     *
     * @param string $name
     * @param array $user
     * @return mixed
     */
    public function create($name, array $user)
    {
        return $this->client->send(sprintf('/api/users/%s', urlencode($name)), 'PUT', [], $user);
    }

    /**
     * Delete a user.
     *
     * @param string $name
     * @return array
     */
    public function delete($name)
    {
        return $this->client->send(sprintf('/api/users/%s', urlencode($name)), 'DELETE');
    }

    /**
     * A list of all permissions for a given user.
     *
     * @param string $name
     * @return array
     */
    public function permissions($name)
    {
        return $this->client->send(sprintf('/api/users/%s/permissions', urlencode($name)));
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Api;

/**
 * Vhost
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Vhost extends AbstractApi
{
    /**
     * A list of all vhosts.
     *
     * @return array
     */
    public function all()
    {
        return $this->client->send('/api/vhosts');
    }

    /**
     * An individual virtual host.
     *
     * @param string $name
     * @return array
     */
    public function get($name)
    {
        return $this->client->send(sprintf('/api/vhosts/%s', urlencode($name)));
    }

    /**
     * As a virtual host only has a name, you do not need an HTTP body when PUTing one of these.
     *
     * @param string $name
     * @return array
     */
    public function create($name)
    {
        return $this->client->send(sprintf('/api/vhosts/%s', urlencode($name)), 'PUT');
    }

    /**
     * Delete a vhost.
     *
     * @param string $name
     * @return array
     */
    public function delete($name)
    {
        return $this->client->send(sprintf('/api/vhosts/%s', urlencode($name)), 'DELETE');
    }

    /**
     * A list of all permissions for a given virtual host.
     *
     * @param string $name
     * @return array
     */
    public function permissions($name)
    {
        return $this->client->send(sprintf('/api/vhosts/%s/permissions', urlencode($name)));
    }
}


code:
<?php

namespace RabbitMq\ManagementApi;

use Http\Client\Common\Plugin\AuthenticationPlugin;
use Http\Client\Common\Plugin\HeaderDefaultsPlugin;
use Http\Client\Common\PluginClient;
use Http\Client\HttpClient;
use Http\Discovery\HttpClientDiscovery;
use Http\Discovery\MessageFactoryDiscovery;
use Http\Message\Authentication\BasicAuth;

/**
 * ManagementApi
 *
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class Client
{
    /**
     * @var HttpClient
     */
    protected $client;
    protected $messageFactory;
    protected $baseUrl;
    protected $username;
    protected $password;

    /**
     * @param HttpClient $client
     * @param string $baseUrl
     * @param string $username
     * @param string $password
     */
    public function __construct(HttpClient $client = null, $baseUrl = 'http://localhost:15672', $username = 'guest', $password = 'guest')
    {
        $this->baseUrl = $baseUrl;
        $this->messageFactory = MessageFactoryDiscovery::find();

        $this->client = new PluginClient(
            $client ?: HttpClientDiscovery::find(), [
            new AuthenticationPlugin(new BasicAuth($username, $password)),
            new HeaderDefaultsPlugin(['Content-Type' => 'application/json'])
        ]);
    }

    /**
     * Declares a test queue, then publishes and consumes a message. Intended for use by monitoring tools. If
     * everything is working correctly, will return HTTP status 200 with body:
     *
     * {"status":"ok"}
     *
     * Note: the test queue will not be deleted (to to prevent queue churn if this is repeatedly pinged).
     *
     * @param string $vhost
     * @return array
     */
    public function alivenessTest($vhost)
    {
        return $this->send(sprintf('/api/aliveness-test/%s', urlencode($vhost)));
    }

    /**
     * Various random bits of information that describe the whole system.
     *
     * @return array
     */
    public function overview()
    {
        return $this->send('/api/overview');
    }

    /**
     * A list of extensions to the management plugin.
     *
     * @return array
     */
    public function extensions()
    {
        return $this->send('/api/extensions');
    }

    /**
     * The server definitions - exchanges, queues, bindings, users, virtual hosts, permissions and parameters.
     *
     * Everything apart from messages. POST to upload an existing set of definitions. Note that:
     *
     * - The definitions are merged. Anything already existing is untouched.
     * - Conflicts will cause an error.
     * - In the event of an error you will be left with a part-applied set of definitions.
     *
     * For convenience you may upload a file from a browser to this URI (i.e. you can use multipart/form-data as well as
     * application/json) in which case the definitions should be uploaded as a form field named "file".
     *
     * @return mixed
     */
    public function definitions()
    {
        return $this->send('/api/definitions');
    }

    /**
     * @return Api\Connection
     */
    public function connections()
    {
        return new Api\Connection($this);
    }

    /**
     * @return Api\Channel
     */
    public function channels()
    {
        return new Api\Channel($this);
    }

    /**
     * @return Api\Exchange
     */
    public function exchanges()
    {
        return new Api\Exchange($this);
    }

    /**
     * @return Api\Queue
     */
    public function queues()
    {
        return new Api\Queue($this);
    }

    /**
     * @return Api\Vhost
     */
    public function vhosts()
    {
        return new Api\Vhost($this);
    }

    /**
     * @return Api\Binding
     */
    public function bindings()
    {
        return new Api\Binding($this);
    }

    /**
     * @return Api\User
     */
    public function users()
    {
        return new Api\User($this);
    }

    /**
     * @return Api\Permission
     */
    public function permissions()
    {
        return new Api\Permission($this);
    }

    /**
     * @return Api\Parameter
     */
    public function parameters()
    {
        return new Api\Parameter($this);
    }

    /**
     * @return Api\Policy
     */
    public function policies()
    {
        return new Api\Policy($this);
    }

    /**
     * @return array
     */
    public function whoami()
    {
        return $this->send('/api/whoami');
    }

    /**
     * @param string $endpoint Resource URI.
     * @param string $method
     * @param array $headers  HTTP headers
     * @param string|resource|array $body Entity body of request (POST/PUT) or response (GET)
     * @return array
     */
    public function send($endpoint, $method = 'GET', array $headers = [], $body = null)
    {
        if (null !== $body) {
            $body = json_encode($body);
        }

        $request = $this->messageFactory->createRequest($method, $this->baseUrl . $endpoint, $headers, $body);
        $response = $this->client->sendRequest($request);

        return json_decode($response->getBody()->getContents(), true);
    }
}


code:
<?php

namespace RabbitMq\ManagementApi\Exception;

/**
 * @author Richard Fullmer <richard.fullmer@opensoftdev.com>
 */
class InvalidArgumentException extends \InvalidArgumentException
{

}


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