Source of file AbstractWorker.php
Size: 14,045 Bytes - Last Modified: 2021-01-12T22:04:13+00:00
C:/Users/MAKS/Code/_PROJECTS/amqp-agent/src/Worker/AbstractWorker.php
| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
Covered by 53 test(s):
98
Covered by 53 test(s):
99
Covered by 53 test(s):
100
Covered by 53 test(s):
101102103104105106107
Covered by 50 test(s):
108
Covered by 50 test(s):
109110111112113114115116117118
Covered by 4 test(s):
119
Covered by 4 test(s):
120
Covered by 3 test(s):
121122123
Covered by 1 test(s):
124125126127128129130131132133134135
Covered by 4 test(s):
136
Covered by 4 test(s):
137138
Covered by 4 test(s):
139
Covered by 3 test(s):
140
Covered by 3 test(s):
141
Covered by 3 test(s):
142
Covered by 3 test(s):
143144145
Covered by 3 test(s):
146147148
Covered by 1 test(s):
149150151152153154155156157158159160
Covered by 7 test(s):
161
Covered by 7 test(s):
162163
Covered by 7 test(s):
164
Covered by 7 test(s):
165166
Covered by 7 test(s):
167
Covered by 7 test(s):
168
Covered by 7 test(s):
169170171
Covered by 6 test(s):
172
Covered by 2 test(s):
173
Covered by 2 test(s):
174175
Covered by 4 test(s):
176
Covered by 4 test(s):
177178179180181
Covered by 1 test(s):
182183184185
Covered by 7 test(s):
186
Covered by 6 test(s):
187188189
Covered by 1 test(s):
190
Covered by 1 test(s):
191
Covered by 1 test(s):
192
Covered by 1 test(s):
193
Covered by 1 test(s):
194
Covered by 1 test(s):
195
Covered by 1 test(s):
196197198199200201202203204205206207
Covered by 1 test(s):
208209210211212213214215216217
Covered by 38 test(s):
218
Covered by 37 test(s):
219220221
Covered by 38 test(s):
222
Covered by 37 test(s):
223224225
Covered by 38 test(s):
226227228229230231232233234
Covered by 51 test(s):
235
Covered by 36 test(s):
236
Covered by 36 test(s):
237238
Covered by 36 test(s):
239
Covered by 36 test(s):
240241242
Covered by 51 test(s):
243
Covered by 36 test(s):
244
Covered by 36 test(s):
245246
Covered by 36 test(s):
247
Covered by 36 test(s):
248249250
Covered by 51 test(s):
251252253254255256257258259
Covered by 1 test(s):
260
Covered by 1 test(s):
261262
Covered by 1 test(s):
263264265266267268269270271272273274
Covered by 20 test(s):
275
Covered by 20 test(s):
276
Covered by 7 test(s):
277278279
Covered by 20 test(s):
280281282
Covered by 20 test(s):
283
Covered by 20 test(s):
284
Covered by 20 test(s):
285
Covered by 20 test(s):
286
Covered by 20 test(s):
287
Covered by 20 test(s):
288
Covered by 20 test(s):
289
Covered by 20 test(s):
290
Covered by 20 test(s):
291292293294295296
Covered by 20 test(s):
297
Covered by 7 test(s):
298299300
Covered by 20 test(s):
301302303304305306307308309310
Covered by 4 test(s):
311312313314315316317318319320321
Covered by 1 test(s):
322
Covered by 1 test(s):
323324325326327328329330331332333
Covered by 37 test(s):
334
Covered by 37 test(s):
335
Covered by 1 test(s):
336337338
Covered by 37 test(s):
339
Covered by 37 test(s):
340
Covered by 37 test(s):
341
Covered by 37 test(s):
342
Covered by 37 test(s):
343
Covered by 37 test(s):
344
Covered by 37 test(s):
345
Covered by 37 test(s):
346
Covered by 37 test(s):
347
Covered by 37 test(s):
348
Covered by 37 test(s):
349
Covered by 37 test(s):
350
Covered by 37 test(s):
351
Covered by 37 test(s):
352
Covered by 37 test(s):
353
Covered by 37 test(s):
354
Covered by 37 test(s):
355356357
Covered by 37 test(s):
358
Covered by 1 test(s):
359360361
Covered by 37 test(s):
362363364365366367368369370
Covered by 11 test(s):
371372373374375376377378379380381
Covered by 1 test(s):
382
Covered by 1 test(s):
383384385386387388389390391392393
Covered by 37 test(s):
394
Covered by 37 test(s):
395
Covered by 2 test(s):
396397398
Covered by 37 test(s):
399400
Covered by 37 test(s):
401
Covered by 37 test(s):
402
Covered by 37 test(s):
403
Covered by 37 test(s):
404405406407
Covered by 37 test(s):
408
Covered by 2 test(s):
409410411
Covered by 37 test(s):
412413414415416417418419420421422
Covered by 2 test(s):
423
Covered by 2 test(s):
424425
Covered by 2 test(s):
426
Covered by 1 test(s):
427428429
Covered by 1 test(s):
430431432
| <?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\Connection\AMQPStreamConnection; use PhpAmqpLib\Channel\AMQPChannel; use PhpAmqpLib\Message\AMQPMessage; use PhpAmqpLib\Wire\AMQPTable; use PhpAmqpLib\Exception\AMQPInvalidArgumentException; use PhpAmqpLib\Exception\AMQPTimeoutException; use PhpAmqpLib\Exception\AMQPConnectionClosedException; use MAKS\AmqpAgent\Worker\AbstractWorkerInterface; use MAKS\AmqpAgent\Worker\WorkerCommandTrait; use MAKS\AmqpAgent\Worker\WorkerMutationTrait; use MAKS\AmqpAgent\Exception\MagicMethodsExceptionsTrait; use MAKS\AmqpAgent\Exception\PropertyDoesNotExistException; use MAKS\AmqpAgent\Exception\AmqpAgentException as Exception; use MAKS\AmqpAgent\Config\AbstractWorkerParameters as Parameters; /** * An abstract class implementing the basic functionality of a worker. * @since 1.0.0 * @api */abstract class AbstractWorker implements AbstractWorkerInterface {use MagicMethodsExceptionsTrait { __get as private __get_MMET; __set as private __set_MMET; } use WorkerMutationTrait; use WorkerCommandTrait; /** * The default connection options that the worker should use when no overrides are provided. * @var array */protected $connectionOptions; /** * The default channel options that the worker should use when no overrides are provided. * @var array */protected $channelOptions; /** * The default queue options that the worker should use when no overrides are provided. * @var array */protected $queueOptions; /** * The default connection of the worker. * @var AMQPStreamConnection */public $connection; /** * The default channel of the worker. * @var AMQPChannel */public $channel; /** * All opened connections of the worker. * @var AMQPStreamConnection[] */public $connections = []; /** * All opened channels of the the worker. * @var AMQPChannel[] */public $channels = []; /** * AbstractWorker 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. */public function __construct( array $connectionOptions = [], array $channelOptions = [], array $queueOptions = [] ) { $this->connectionOptions = Parameters::patch($connectionOptions, 'CONNECTION_OPTIONS'); $this->channelOptions = Parameters::patch($channelOptions, 'CHANNEL_OPTIONS'); $this->queueOptions = Parameters::patch($queueOptions, 'QUEUE_OPTIONS'); } /** * Closes the connection with RabbitMQ server before destroying the object. */public function __destruct() { $this->disconnect(); } /** * Gets a class member via public property access notation. * @param string $member Property name. * @return mixed * @throws PropertyDoesNotExistException */public function __get(string $member) { $isMember = property_exists($this, $member); if ($isMember) { return $this->{$member}; } $this->__get_MMET($member); } /** * Sets a class member via public property assignment notation. * @param string $member Property name. * @param array $array Array of overrides. The array type here is important, because only *Options properties should be overridable. * @return void * @throws PropertyDoesNotExistException */public function __set(string $member, array $array) { $isMember = property_exists($this, $member); $notProtected = $member !== 'mutation'; if ($isMember && $notProtected) { $acceptedKeys = array_keys($this->{$member}); foreach ($array as $key => $value) { if (in_array($key, $acceptedKeys)) { $this->{$member}[$key] = $value; } } return; } $this->__set_MMET($member, $array); } /** * Closes the connection or the channel or both with RabbitMQ server. * @param AMQPStreamConnection|AMQPChannel|AMQPMessage ...$object The object that should be used to close the channel or the connection. * @return bool True on success. * @throws Exception */public static function shutdown(...$object): bool { $successful = true; $parameters = []; foreach ($object as $class) { $parameters[] = is_object($class) ? get_class($class) : gettype($class); if ( $class instanceof AMQPStreamConnection || $class instanceof AMQPChannel || $class instanceof AMQPMessage ) { try { if (!($class instanceof AMQPMessage)) { $class->close(); continue; } $class->getChannel()->close(); } catch (AMQPConnectionClosedException $e) { // No need to throw the exception here as it's extraneous. This error // happens when a channel gets closed multiple times in different ways. } } else { $successful = false; } } if ($successful) { return $successful; } throw new Exception( sprintf( 'The passed parameter must be of type %s, %s or %s or a combination of them. Given parameter(s) has/have the type(s): %s!', AMQPStreamConnection::class, AMQPChannel::class, AMQPMessage::class, implode(', ', $parameters) ) ); } /** * Returns an AMQPTable object. * @param array $array An array of the option wished to be turn into the an arguments object. * @return AMQPTable */public static function arguments(array $array): AMQPTable { return new AMQPTable($array); } /** * Establishes a connection with RabbitMQ server and opens a channel for the worker in the opened connection, it also sets both of them as defaults. * @return self */public function connect() { if (empty($this->connection)) { $this->connection = $this->getNewConnection(); } if (empty($this->channel)) { $this->channel = $this->getNewChannel(); } return $this; } /** * Closes all open channels and connections with RabbitMQ server. * @return self */public function disconnect() { if (count($this->channels)) { foreach ($this->channels as $channel) { $channel->close(); } $this->channel = null; $this->channels = []; } if (count($this->connections)) { foreach ($this->connections as $connection) { $connection->close(); } $this->connection = null; $this->connections = []; } return $this; } /** * Executes `self::disconnect()` and `self::connect()` respectively. Note that this method will not restore old channels. * @return self */public function reconnect() { $this->disconnect(); $this->connect(); return $this; } /** * Declares a queue on the default channel of the worker's connection with RabbitMQ server. * @param array $parameters [optional] The overrides for the default queue options of the worker. * @param AMQPChannel $_channel [optional] The channel that should be used instead of the default worker's channel. * @return self * @throws AMQPTimeoutException */public function queue(?array $parameters = null, ?AMQPChannel $_channel = null) { $changes = null; if ($parameters) { $changes = $this->mutateClassMember('queueOptions', $parameters); } $channel = $_channel ?: $this->channel; try { $channel->queue_declare( $this->queueOptions['queue'], $this->queueOptions['passive'], $this->queueOptions['durable'], $this->queueOptions['exclusive'], $this->queueOptions['auto_delete'], $this->queueOptions['nowait'], $this->queueOptions['arguments'], $this->queueOptions['ticket'] ); } catch (AMQPTimeoutException $error) { // @codeCoverageIgnore Exception::rethrow($error); // @codeCoverageIgnore } if ($changes) { $this->mutateClassMember('queueOptions', $changes); } return $this; } /** * Returns the default connection of the worker. If the worker is not connected, it returns null. * @since 1.1.0 * @return AMQPStreamConnection|null */public function getConnection(): ?AMQPStreamConnection { return $this->connection; } /** * Sets the passed connection as the default connection of the worker. * @since 1.1.0 * @param AMQPStreamConnection $connection The connection that should be as the default connection of the worker. * @return self */public function setConnection(AMQPStreamConnection $connection) { $this->connection = $connection; return $this; } /** * Opens a new connection to RabbitMQ server and returns it. Connections returned by this method pushed to connections array and are not set as default automatically. * @since 1.1.0 * @param array|null $parameters * @return AMQPStreamConnection */public function getNewConnection(array $parameters = null): AMQPStreamConnection { $changes = null; if ($parameters) { $changes = $this->mutateClassMember('connectionOptions', $parameters); } $this->connections[] = $connection = new AMQPStreamConnection( $this->connectionOptions['host'], $this->connectionOptions['port'], $this->connectionOptions['user'], $this->connectionOptions['password'], $this->connectionOptions['vhost'], $this->connectionOptions['insist'], $this->connectionOptions['login_method'], $this->connectionOptions['login_response'], $this->connectionOptions['locale'], $this->connectionOptions['connection_timeout'], $this->connectionOptions['read_write_timeout'], $this->connectionOptions['context'], $this->connectionOptions['keepalive'], $this->connectionOptions['heartbeat'], $this->connectionOptions['channel_rpc_timeout'], $this->connectionOptions['ssl_protocol'] ); if ($changes) { $this->mutateClassMember('connectionOptions', $changes); } return $connection; } /** * Returns the default channel of the worker. If the worker is not connected, it returns null. * @return AMQPChannel|null */public function getChannel(): ?AMQPChannel { return $this->channel; } /** * Sets the passed channel as the default channel of the worker. * @since 1.1.0 * @param AMQPChannel $channel The channel that should be as the default channel of the worker. * @return self */public function setChannel(AMQPChannel $channel) { $this->channel = $channel; return $this; } /** * Returns a new channel on the the passed connection of the worker. If no connection is passed, it uses the default connection. If the worker is not connected, it returns null. * @param array|null $parameters [optional] The overrides for the default channel options of the worker. * @param AMQPStreamConnection|null $_connection [optional] The connection that should be used instead of the default worker's connection. * @return AMQPChannel|null */public function getNewChannel(array $parameters = null, ?AMQPStreamConnection $_connection = null): ?AMQPChannel { $changes = null; if ($parameters) { $changes = $this->mutateClassMember('channelOptions', $parameters); } $connection = $_connection ?: $this->connection; $channel = null; if (isset($connection)) { $this->channels[] = $channel = $connection->channel( $this->channelOptions['channel_id'] ); } if ($changes) { $this->mutateClassMember('channelOptions', $changes); } return $channel; } /** * Fetches a channel object identified by the passed id (channel_id). If not found, it returns null. * @param int $channelId The id of the channel wished to be fetched. * @param AMQPStreamConnection|null $_connection [optional] The connection that should be used instead of the default worker's connection. * @return AMQPChannel|null */public function getChannelById(int $channelId, ?AMQPStreamConnection $_connection = null): ?AMQPChannel { $connection = $_connection ?: $this->connection; $channels = $connection->channels; if (array_key_exists($channelId, $channels)) { return $channels[$channelId]; } return null; } } |