Source of file Publisher.php

<?php

/**

 * @author Marwan Al-Soltany <MarwanAlsoltany@gmail.com>

 * @copyright Marwan Al-Soltany 2020

 * For the full copyright and license information, please view

 * the LICENSE file that was distributed with this source code.

 */

declare(strict_types=1);

namespace MAKS\AmqpAgent\Worker;

use PhpAmqpLib\Channel\AMQPChannel;

use PhpAmqpLib\Message\AMQPMessage;

use PhpAmqpLib\Exception\AMQPInvalidArgumentException;

use PhpAmqpLib\Exception\AMQPTimeoutException;

use PhpAmqpLib\Exception\AMQPConnectionBlockedException;

use PhpAmqpLib\Exception\AMQPConnectionClosedException;

use PhpAmqpLib\Exception\AMQPChannelClosedException;

use MAKS\AmqpAgent\Worker\AbstractWorker;

use MAKS\AmqpAgent\Worker\PublisherInterface;

use MAKS\AmqpAgent\Worker\WorkerFacilitationInterface;

use MAKS\AmqpAgent\Exception\AmqpAgentException as Exception;

use MAKS\AmqpAgent\Config\PublisherParameters as Parameters;

/**

 * A class specialized in publishing. Implementing only the methods needed for a publisher.

 *

 * Example:

 * ```

 * $publisher = new Publisher();

 * $publisher->connect();

 * $publisher->queue();

 * $publisher->exchange();

 * $publisher->bind();

 * $publisher->publish('Some message!');

 * $publisher->disconnect();

 * ```

 *

 * @since 1.0.0

 * @api

 */

class Publisher extends AbstractWorker implements PublisherInterface, WorkerFacilitationInterface

{

    /**

     * The default exchange options that the worker should use when no overrides are provided.

     * @var array

     */

    protected $exchangeOptions;

    /**

     * The default bind options that the worker should use when no overrides are provided.

     * @var array

     */

    protected $bindOptions;

    /**

     * The default message options that the worker should use when no overrides are provided.

     * @var array

     */

    protected $messageOptions;

    /**

     * The default publish options that the worker should use when no overrides are provided.

     * @var array

     */

    protected $publishOptions;

    /**

     * Publisher object constructor.

     * @param array $connectionOptions [optional] The overrides for the default connection options of the worker.

     * @param array $channelOptions [optional] The overrides for the default channel options of the worker.

     * @param array $queueOptions [optional] The overrides for the default queue options of the worker.

     * @param array $exchangeOptions [optional] The overrides for the default exchange options of the worker.

     * @param array $bindOptions [optional] The overrides for the default bind options of the worker.

     * @param array $messageOptions [optional] The overrides for the default message options of the worker.

     * @param array $publishOptions [optional] The overrides for the default publish options of the worker.

     */

    public function __construct(

        array $connectionOptions = [],

        array $channelOptions = [],

        array $queueOptions = [],

        array $exchangeOptions = [],

        array $bindOptions = [],

        array $messageOptions = [],

        array $publishOptions = []

    ) {

        $this->exchangeOptions = Parameters::patch($exchangeOptions, 'EXCHANGE_OPTIONS');

        $this->bindOptions     = Parameters::patch($bindOptions, 'BIND_OPTIONS');

        $this->messageOptions  = Parameters::patch($messageOptions, 'MESSAGE_OPTIONS');

        $this->publishOptions  = Parameters::patch($publishOptions, 'PUBLISH_OPTIONS');

        parent::__construct($connectionOptions, $channelOptions, $queueOptions);

    }

    /**

     * Declares an exchange on the default channel of the worker's connection to RabbitMQ server.

     * @param array|null $parameters [optional] The overrides for the default exchange options of the worker.

     * @param AMQPChannel|null $_channel [optional] The channel that should be used instead of the default worker's channel.

     * @return self

     * @throws AMQPTimeoutException

     */

    public function exchange(?array $parameters = null, ?AMQPChannel $_channel = null)

    {

        $changes = null;

        if ($parameters) {

            $changes = $this->mutateClassMember('exchangeOptions', $parameters);

        }

        $channel = $_channel ?: $this->channel;

        try {

            $channel->exchange_declare(

                $this->exchangeOptions['exchange'],

                $this->exchangeOptions['type'],

                $this->exchangeOptions['passive'],

                $this->exchangeOptions['durable'],

                $this->exchangeOptions['auto_delete'],

                $this->exchangeOptions['internal'],

                $this->exchangeOptions['nowait'],

                $this->exchangeOptions['arguments'],

                $this->exchangeOptions['ticket']

            );

        } catch (AMQPTimeoutException $error) { // @codeCoverageIgnore

            Exception::rethrow($error); // @codeCoverageIgnore

        }

        if ($changes) {

            $this->mutateClassMember('exchangeOptions', $changes);

        }

        return $this;

    }

    /**

     * Binds the default queue to the default exchange on the default channel of the worker's connection to RabbitMQ server.

     * @param array|null $parameters [optional] The overrides for the default bind options of the worker.

     * @param AMQPChannel|null $_channel [optional] The channel that should be used instead of the default worker's channel.

     * @return self

     * @throws AMQPTimeoutException

     */

    public function bind(?array $parameters = null, ?AMQPChannel $_channel = null)

    {

        $changes = null;

        if ($parameters) {

            $changes = $this->mutateClassMember('bindOptions', $parameters);

        }

        $channel = $_channel ?: $this->channel;

        try {

            $channel->queue_bind(

                $this->bindOptions['queue'],

                $this->bindOptions['exchange'],

                $this->bindOptions['routing_key'],

                $this->bindOptions['nowait'],

                $this->bindOptions['arguments'],

                $this->bindOptions['ticket']

            );

        } catch (AMQPTimeoutException $error) { // @codeCoverageIgnore

            Exception::rethrow($error); // @codeCoverageIgnore

        }

        if ($changes) {

            $this->mutateClassMember('bindOptions', $changes);

        }

        return $this;

    }

    /**

     * Returns an AMQPMessage object.

     * @param string $body The body of the message.

     * @param array|null $properties [optional] The overrides for the default properties of the default message options of the worker.

     * @return AMQPMessage

     */

    public function message(string $body, ?array $properties = null): AMQPMessage

    {

        $changes = null;

        if ($properties) {

            $changes = $this->mutateClassSubMember('messageOptions', 'properties', $properties);

        }

        if ($body) {

            $this->messageOptions['body'] = $body;

        }

        $message = new AMQPMessage(

            $this->messageOptions['body'],

            $this->messageOptions['properties']

        );

        if ($changes) {

            $this->mutateClassSubMember('messageOptions', 'properties', $changes);

        }

        return $message;

    }

    /**

     * Publishes a message to the default exchange on the default channel of the worker's connection to RabbitMQ server.

     * @param string|array|AMQPMessage $payload A string of the body of the message or an array of body and properties for the message or a AMQPMessage object.

     * @param array|null $parameters [optional] The overrides for the default publish options of the worker.

     * @param AMQPChannel|null $_channel [optional] The channel that should be used instead of the default worker's channel.

     * @return self

     * @throws Exception|AMQPChannelClosedException|AMQPConnectionClosedException|AMQPConnectionBlockedException

     */

    public function publish($payload, ?array $parameters = null, ?AMQPChannel $_channel = null)

    {

        $changes = null;

        if ($parameters) {

            $changes = $this->mutateClassMember('publishOptions', $parameters);

        }

        $channel = $_channel ?: $this->channel;

        $originalMessage = $this->publishOptions['msg'];

        $message = $payload ?: $originalMessage;

        if ($message instanceof AMQPMessage) {

            $this->publishOptions['msg'] = $message;

        } elseif (is_array($message) && isset($message['body']) && isset($message['properties'])) {

            $this->publishOptions['msg'] = $this->message($message['body'], $message['properties']);

        } elseif (is_string($message)) {

            $this->publishOptions['msg'] = $this->message($message);

        } else {

            throw new Exception(

                sprintf(

                    'Payload must be a string, an array like %s, or an instance of "%s". The given parameter (data-type: %s) was none of them.',

                    '["body" => "Message body!", "properties" ["key" => "value"]]',

                    AMQPMessage::class,

                    is_object($payload) ? get_class($payload) : gettype($payload)

                )

            );

        }

        try {

            $channel->basic_publish(

                $this->publishOptions['msg'],

                $this->publishOptions['exchange'],

                $this->publishOptions['routing_key'],

                $this->publishOptions['mandatory'],

                $this->publishOptions['immediate'],

                $this->publishOptions['ticket']

            );

        } catch (AMQPChannelClosedException | AMQPConnectionClosedException | AMQPConnectionBlockedException $error) { // @codeCoverageIgnore

            Exception::rethrow($error); // @codeCoverageIgnore

        } finally {

            // reverting messageOptions back to its state.

            $this->publishOptions['msg'] = $originalMessage;

        }

        if ($changes) {

            $this->mutateClassMember('publishOptions', $changes);

        }

        return $this;

    }

    /**

     * Publishes a batch of messages to the default exchange on the default channel of the worker's connection to RabbitMQ server.

     * @param string[]|array[]|AMQPMessage[] $messages An array of bodies of the messages or an array of arrays of body and properties for the messages or an array of AMQPMessage objects.

     * @param int $batchSize [optional] The number of messages that should be published per batch.

     * @param array|null $parameters [optional] The overrides for the default publish options of the worker.

     * @param AMQPChannel|null $_channel [optional] The channel that should be used instead of the default worker's channel.

     * @return self

     * @throws Exception|AMQPChannelClosedException|AMQPConnectionClosedException|AMQPConnectionBlockedException

     */

    public function publishBatch(array $messages, int $batchSize = 2500, ?array $parameters = null, ?AMQPChannel $_channel = null)

    {

        $changes = null;

        if ($parameters) {

            $changes = $this->mutateClassMember('publishOptions', $parameters);

        }

        $channel = $_channel ?: $this->channel;

        $originalMessage = $this->publishOptions['msg'];

        $count = count($messages);

        for ($i = 0; $i < $count; $i++) {

            $payload = $messages[$i];

            $message = $payload ?: $originalMessage;

            if ($message instanceof AMQPMessage) {

                $this->publishOptions['msg'] = $message;

            } elseif (is_array($message) && isset($message['body']) && isset($message['properties'])) {

                $this->publishOptions['msg'] = $this->message($message['body'], $message['properties']);

            } elseif (is_string($message)) {

                $this->publishOptions['msg'] = $this->message($message);

            } else {

                throw new Exception(

                    sprintf(

                        'Messages array elements must be either a string, an array like %s, or an instance of "%s". Element in index "%d" (data-type: %s) was none of them.',

                        '["body" => "Message body!", "properties" ["key" => "value"]]',

                        AMQPMessage::class,

                        $i,

                        is_object($payload) ? get_class($payload) : gettype($payload)

                    )

                );

            }

            $channel->batch_basic_publish(

                $this->publishOptions['msg'],

                $this->publishOptions['exchange'],

                $this->publishOptions['routing_key'],

                $this->publishOptions['mandatory'],

                $this->publishOptions['immediate'],

                $this->publishOptions['ticket']

            );

            if ($i % $batchSize == 0) {

                try {

                    $channel->publish_batch();

                    // @codeCoverageIgnoreStart

                } catch (AMQPConnectionBlockedException $e) {

                    $tries = -1;

                    do {

                        sleep(1);

                        $tries++;

                    } while ($this->connection->isBlocked() && $tries >= 60);

                    $channel->publish_batch();

                } catch (AMQPChannelClosedException | AMQPConnectionClosedException | AMQPConnectionBlockedException $error) {

                    Exception::rethrow($error);

                    // @codeCoverageIgnoreEnd

                }

            }

        }

        try {

            $channel->publish_batch();

        } catch (AMQPChannelClosedException | AMQPConnectionClosedException | AMQPConnectionBlockedException $error) { // @codeCoverageIgnore

            Exception::rethrow($error); // @codeCoverageIgnore

        } finally {

            // reverting messageOptions back to its state.

            $this->publishOptions['msg'] = $originalMessage;

        }

        if ($changes) {

            $this->mutateClassMember('publishOptions', $changes);

        }

        return $this;

    }

    /**

     * Executes `self::connect()`, `self::queue()`, `self::exchange`, and `self::bind()` respectively.

     * @return self

     */

    public function prepare()

    {

        $this->connect();

        $this->queue();

        $this->exchange();

        $this->bind();

        return $this;

    }

    /**

     * Executes `self::connect()`, `self::queue()`, `self::exchange`, `self::bind()`, `self::publish()`, and `self::disconnect()` respectively.

     * @param string[]|array[]|AMQPMessage[] $messages An array of strings, arrays, or AMQPMessage objects (same as `self::publishBatch()`).

     * @return void

     * @throws Exception

     */

    public function work($messages): void

    {

        try {

            $this->prepare();

            foreach ($messages as $message) {

                $this->publish($message);

            }

            $this->disconnect();

        } catch (Exception $error) {

            Exception::rethrow($error, null, false);

        }

    }

}