From 6a1a2c4734e4518f015aa1993d0aa2da63ce5579 Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Thu, 29 Jun 2023 16:18:04 +0300 Subject: [PATCH] Push notifications support (#1316) * Added new push notifications consumer * Added push notifications dispatcher loop, intoduced consumer and dispatcher interfaces * Static analysis fixes * Codestyle fix * Codestyle fixes * Moved RelayConsumer in appropriate directory, fixed access modifiers --- examples/dispatcher_loop.php | 2 +- examples/push_notifications.php | 49 +++++ examples/push_notifications_dispatcher.php | 68 ++++++ src/Client.php | 16 +- src/ClientContextInterface.php | 6 +- src/ClientInterface.php | 6 +- src/Connection/StreamConnection.php | 12 +- src/Consumer/AbstractConsumer.php | 112 ++++++++++ src/Consumer/AbstractDispatcherLoop.php | 86 ++++++++ src/Consumer/ConsumerInterface.php | 64 ++++++ src/Consumer/DispatcherLoopInterface.php | 65 ++++++ .../PubSub/Consumer.php} | 198 ++++++++++++------ src/Consumer/PubSub/DispatcherLoop.php | 108 ++++++++++ src/{ => Consumer}/PubSub/RelayConsumer.php | 12 +- src/Consumer/Push/Consumer.php | 55 +++++ src/Consumer/Push/DispatcherLoop.php | 43 ++++ .../Push/PushNotificationException.php | 19 ++ src/Consumer/Push/PushResponse.php | 71 +++++++ src/Consumer/Push/PushResponseInterface.php | 37 ++++ src/PubSub/Consumer.php | 157 -------------- src/PubSub/DispatcherLoop.php | 171 --------------- tests/Predis/ClientTest.php | 8 +- .../Predis/Consumer/AbstractConsumerTest.php | 134 ++++++++++++ .../Consumer/AbstractDispatcherLoopTest.php | 130 ++++++++++++ .../{ => Consumer}/PubSub/ConsumerTest.php | 5 +- .../PubSub/DispatcherLoopTest.php | 2 +- tests/Predis/Consumer/Push/ConsumerTest.php | 88 ++++++++ 27 files changed, 1309 insertions(+), 415 deletions(-) create mode 100644 examples/push_notifications.php create mode 100644 examples/push_notifications_dispatcher.php create mode 100644 src/Consumer/AbstractConsumer.php create mode 100644 src/Consumer/AbstractDispatcherLoop.php create mode 100644 src/Consumer/ConsumerInterface.php create mode 100644 src/Consumer/DispatcherLoopInterface.php rename src/{PubSub/AbstractConsumer.php => Consumer/PubSub/Consumer.php} (50%) create mode 100644 src/Consumer/PubSub/DispatcherLoop.php rename src/{ => Consumer}/PubSub/RelayConsumer.php (94%) create mode 100644 src/Consumer/Push/Consumer.php create mode 100644 src/Consumer/Push/DispatcherLoop.php create mode 100644 src/Consumer/Push/PushNotificationException.php create mode 100644 src/Consumer/Push/PushResponse.php create mode 100644 src/Consumer/Push/PushResponseInterface.php delete mode 100644 src/PubSub/Consumer.php delete mode 100644 src/PubSub/DispatcherLoop.php create mode 100644 tests/Predis/Consumer/AbstractConsumerTest.php create mode 100644 tests/Predis/Consumer/AbstractDispatcherLoopTest.php rename tests/Predis/{ => Consumer}/PubSub/ConsumerTest.php (99%) rename tests/Predis/{ => Consumer}/PubSub/DispatcherLoopTest.php (99%) create mode 100644 tests/Predis/Consumer/Push/ConsumerTest.php diff --git a/examples/dispatcher_loop.php b/examples/dispatcher_loop.php index 92f3796c..9dd4fcc3 100644 --- a/examples/dispatcher_loop.php +++ b/examples/dispatcher_loop.php @@ -30,7 +30,7 @@ $client = new Predis\Client($single_server + ['read_write_timeout' => 0]); $pubsub = $client->pubSubLoop(); // Create a dispatcher loop instance and attach a bunch of callbacks. -$dispatcher = new Predis\PubSub\DispatcherLoop($pubsub); +$dispatcher = new \Predis\Consumer\PubSub\DispatcherLoop($pubsub); // Demonstrate how to use a callable class as a callback for the dispatcher loop. class EventsListener implements Countable diff --git a/examples/push_notifications.php b/examples/push_notifications.php new file mode 100644 index 00000000..c3c0fd16 --- /dev/null +++ b/examples/push_notifications.php @@ -0,0 +1,49 @@ + 0, 'protocol' => 3]); + +// 2. Create push notifications consumer. Provides callback where current consumer subscribes to few channels before enter the loop. +$push = $client->push(static function (ClientInterface $client) { + $response = $client->subscribe('channel', 'control'); + $status = ($response[2] === 1) ? 'OK' : 'FAILED'; + echo "Channel subscription status: {$status}\n"; +}); + +// 3. Run consumer that will handle message data type push notifications. And stops if certain message will be sent to control channel. +// Send following commands via redis-cli to test: +// +// PUBLISH channel message1 +// PUBLISH channel message2 +// PUBLISH channel message3 +// PUBLISH control terminate +// Data types should be changed in near future. Instead of Message data type it should be one of kind data types. + +foreach ($push as $notification) { + if ((null !== $notification) && $notification->getDataType() === PushResponseInterface::MESSAGE_DATA_TYPE) { + if ($notification[1] === 'control' && $notification[2] === 'terminate') { + echo "Terminating notification consumer.\n"; + $push->stop(); + break; + } + + $message = $notification[2]; + + echo "Received message: {$message}\n"; + } +} diff --git a/examples/push_notifications_dispatcher.php b/examples/push_notifications_dispatcher.php new file mode 100644 index 00000000..5457cd17 --- /dev/null +++ b/examples/push_notifications_dispatcher.php @@ -0,0 +1,68 @@ + 0, 'protocol' => 3]); + +// 2. Create push notifications consumer. Provides callback where current consumer subscribes to few channels before enter the loop. +$push = $client->push(static function (ClientInterface $client) { + $response = $client->subscribe('channel', 'control'); + $status = ($response[2] === 1) ? 'OK' : 'FAILED'; + echo "Channel subscription status: {$status}\n"; +}); + +// 3. Storage for upcoming notifications. +$messages = []; + +// 4. Create dispatcher for push notifications. +$dispatcher = new DispatcherLoop($push); + +// 5. Attach callback for message data type. Print every message and store them in storage. +// Send following commands via redis-cli to test: +// +// PUBLISH channel message1 +// PUBLISH channel message2 +// PUBLISH channel message3 +// PUBLISH control terminate +// Data types should be changed in near future. Instead of Message data type it should be one of kind data types. + +$dispatcher->attachCallback( + PushResponseInterface::MESSAGE_DATA_TYPE, + static function (array $payload, DispatcherLoopInterface $dispatcher) { + global $messages; + [$channel, $message] = $payload; + + if ($channel === 'control' && $message === 'terminate') { + echo "Terminating notification consumer.\n"; + $dispatcher->stop(); + + return; + } + + $messages[] = $message; + echo "Received message: {$message}\n"; + } +); + +// 6. Run consumer loop with attached callbacks. +$dispatcher->run(); + +// 7. Count all messages that were received during consumer loop. +$messagesCount = count($messages); +echo "We received: {$messagesCount} messages\n"; diff --git a/src/Client.php b/src/Client.php index f2d45628..a8b46150 100644 --- a/src/Client.php +++ b/src/Client.php @@ -26,14 +26,15 @@ use Predis\Connection\ConnectionInterface; use Predis\Connection\Parameters; use Predis\Connection\ParametersInterface; use Predis\Connection\RelayConnection; +use Predis\Consumer\PubSub\Consumer as PubSubConsumer; +use Predis\Consumer\PubSub\RelayConsumer as RelayPubSubConsumer; +use Predis\Consumer\Push\Consumer as PushConsumer; use Predis\Monitor\Consumer as MonitorConsumer; use Predis\Pipeline\Atomic; use Predis\Pipeline\FireAndForget; use Predis\Pipeline\Pipeline; use Predis\Pipeline\RelayAtomic; use Predis\Pipeline\RelayPipeline; -use Predis\PubSub\Consumer as PubSubConsumer; -use Predis\PubSub\RelayConsumer as RelayPubSubConsumer; use Predis\Response\ErrorInterface as ErrorResponseInterface; use Predis\Response\ResponseInterface; use Predis\Response\ServerException; @@ -554,6 +555,17 @@ class Client implements ClientInterface, IteratorAggregate return $this->sharedContextFactory('createPubSub', func_get_args()); } + /** + * Creates new push notifications consumer. + * + * @param callable|null $preLoopCallback Callback that should be called on client before enter a loop. + * @return PushConsumer + */ + public function push(callable $preLoopCallback = null): PushConsumer + { + return new PushConsumer($this, $preLoopCallback); + } + /** * Actual publish/subscribe context initializer method. * diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php index 95f1590e..d76a59ca 100644 --- a/src/ClientContextInterface.php +++ b/src/ClientContextInterface.php @@ -225,6 +225,9 @@ use Predis\Command\Redis\Container\Search\FTCURSOR; * @method $this srandmember($key, $count = null) * @method $this srem($key, $member) * @method $this sscan($key, $cursor, array $options = null) + * @method $this ssubscribe(string ...$shardChannels) + * @method $this subscribe(string ...$channels) + * @method $this sunsubscribe(?string ...$shardChannels = null) * @method $this sunion(array|string $keys) * @method $this sunionstore($destination, array|string $keys) * @method $this tdigestadd(string $key, float ...$value) @@ -304,6 +307,7 @@ use Predis\Command\Redis\Container\Search\FTCURSOR; * @method $this exec() * @method $this multi() * @method $this unwatch() + * @method $this unsubscribe(string ...$channels) * @method $this watch($key) * @method $this eval($script, $numkeys, $keyOrArg1 = null, $keyOrArgN = null) * @method $this eval_ro(string $script, array $keys, ...$argument) @@ -316,7 +320,7 @@ use Predis\Command\Redis\Container\Search\FTCURSOR; * @method $this select($database) * @method $this bgrewriteaof() * @method $this bgsave() - * @method $this client($subcommand, $argument = null) + * @method $this client($subcommand, ...$argument = null) * @method $this config($subcommand, $argument = null) * @method $this dbsize() * @method $this flushall() diff --git a/src/ClientInterface.php b/src/ClientInterface.php index 924db1ec..143b1e9e 100644 --- a/src/ClientInterface.php +++ b/src/ClientInterface.php @@ -234,8 +234,11 @@ use Predis\Response\Status; * @method string|null srandmember(string $key, int $count = null) * @method int srem(string $key, array|string $member) * @method array sscan(string $key, int $cursor, array $options = null) + * @method array ssubscribe(string ...$shardChannels) + * @method array subscribe(string ...$channels) * @method string[] sunion(array|string $keys) * @method int sunionstore(string $destination, array|string $keys) + * @method array sunsubscribe(?string ...$shardChannels = null) * @method int touch(string[]|string $keyOrKeys, string ...$keys = null) * @method Status tdigestadd(string $key, float ...$value) * @method array tdigestbyrank(string $key, int ...$rank) @@ -322,6 +325,7 @@ use Predis\Response\Status; * @method array|null exec() * @method mixed multi() * @method mixed unwatch() + * @method array unsubscribe(string ...$channels) * @method mixed watch(string $key) * @method mixed eval(string $script, int $numkeys, string ...$keyOrArg = null) * @method mixed eval_ro(string $script, array $keys, ...$argument) @@ -334,7 +338,7 @@ use Predis\Response\Status; * @method mixed select(int $database) * @method mixed bgrewriteaof() * @method mixed bgsave() - * @method mixed client($subcommand, $argument = null) + * @method mixed client($subcommand, ...$argument = null) * @method mixed config($subcommand, $argument = null) * @method int dbsize() * @method mixed flushall() diff --git a/src/Connection/StreamConnection.php b/src/Connection/StreamConnection.php index 620a327c..8614de1f 100644 --- a/src/Connection/StreamConnection.php +++ b/src/Connection/StreamConnection.php @@ -14,6 +14,8 @@ namespace Predis\Connection; use InvalidArgumentException; use Predis\Command\CommandInterface; +use Predis\Consumer\Push\PushNotificationException; +use Predis\Consumer\Push\PushResponse; use Predis\Protocol\Parser\UnexpectedTypeException; use Predis\Response\Error; use Predis\Response\ErrorInterface as ErrorResponseInterface; @@ -279,6 +281,7 @@ class StreamConnection extends AbstractConnection /** * {@inheritdoc} + * @throws PushNotificationException */ public function read() { @@ -303,6 +306,13 @@ class StreamConnection extends AbstractConnection switch ($parsedData['type']) { case 'push': + $data = []; + + for ($i = 0; $i < $parsedData['value']; ++$i) { + $data[$i] = $this->read(); + } + + return new PushResponse($data); case 'array': $data = []; @@ -374,7 +384,7 @@ class StreamConnection extends AbstractConnection /** * Reads given resource split on chunks with given size. * - * @param $resource + * @param $resource * @param int $chunkSize * @return string */ diff --git a/src/Consumer/AbstractConsumer.php b/src/Consumer/AbstractConsumer.php new file mode 100644 index 00000000..9644369d --- /dev/null +++ b/src/Consumer/AbstractConsumer.php @@ -0,0 +1,112 @@ +client = $client; + } + + /** + * {@inheritDoc} + */ + public function stop(bool $drop = false): bool + { + $this->isValid = false; + + if ($drop) { + $this->client->disconnect(); + + return true; + } + + return true; + } + + public function getClient(): ClientInterface + { + return $this->client; + } + + /** + * {@inheritDoc} + */ + public function current() + { + return $this->getValue(); + } + + /** + * Returns last message from server. + * + * @return mixed + */ + #[ReturnTypeWillChange] + abstract protected function getValue(); + + /** + * {@inheritDoc} + */ + public function valid() + { + return $this->isValid; + } + + /** + * {@inheritDoc} + */ + public function next() + { + if ($this->valid()) { + ++$this->position; + } + } + + /** + * {@inheritDoc} + */ + #[ReturnTypeWillChange] + public function key() + { + return $this->position; + } + + /** + * {@inheritDoc} + */ + #[ReturnTypeWillChange] + public function rewind() + { + // NOOP + } +} diff --git a/src/Consumer/AbstractDispatcherLoop.php b/src/Consumer/AbstractDispatcherLoop.php new file mode 100644 index 00000000..6b82e5de --- /dev/null +++ b/src/Consumer/AbstractDispatcherLoop.php @@ -0,0 +1,86 @@ +consumer = $consumer; + } + + /** + * {@inheritDoc} + */ + public function getConsumer(): ConsumerInterface + { + return $this->consumer; + } + + /** + * {@inheritDoc} + */ + public function setDefaultCallback(callable $callback = null): void + { + $this->defaultCallback = $callback; + } + + /** + * {@inheritDoc} + */ + public function attachCallback(string $messageType, callable $callback): void + { + $this->callbacksDictionary[$messageType] = $callback; + } + + /** + * {@inheritDoc} + */ + public function detachCallback(string $messageType): void + { + if (isset($this->callbacksDictionary[$messageType])) { + unset($this->callbacksDictionary[$messageType]); + } + } + + /** + * {@inheritDoc} + */ + abstract public function run(): void; + + /** + * {@inheritDoc} + */ + public function stop(): void + { + $this->consumer->stop(); + } +} diff --git a/src/Consumer/ConsumerInterface.php b/src/Consumer/ConsumerInterface.php new file mode 100644 index 00000000..0639c050 --- /dev/null +++ b/src/Consumer/ConsumerInterface.php @@ -0,0 +1,64 @@ +checkCapabilities($client); + + $this->options = $options ?: []; + $this->client = $client; + + $this->genericSubscribeInit('subscribe'); + $this->genericSubscribeInit('psubscribe'); + } + + /** + * Checks if the client instance satisfies the required conditions needed to + * initialize a PUB/SUB consumer. + * + * @param ClientInterface $client Client instance used by the consumer. + * + * @throws NotSupportedException + */ + private function checkCapabilities(ClientInterface $client) + { + if ($client->getConnection() instanceof ClusterInterface) { + throw new NotSupportedException( + 'Cannot initialize a PUB/SUB consumer over cluster connections.' + ); + } + + $commands = ['publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe']; + + if (!$client->getCommandFactory()->supports(...$commands)) { + throw new NotSupportedException( + 'PUB/SUB commands are not supported by the current command factory.' + ); + } + } + + /** + * This method shares the logic to handle both SUBSCRIBE and PSUBSCRIBE. + * + * @param string $subscribeAction Type of subscription. + */ + private function genericSubscribeInit($subscribeAction) + { + if (isset($this->options[$subscribeAction])) { + $this->$subscribeAction($this->options[$subscribeAction]); + } + } + + /** + * {@inheritdoc} + */ + protected function writeRequest($method, $arguments) + { + $this->client->getConnection()->writeRequest( + $this->client->createCommand($method, + Command::normalizeArguments($arguments) + ) + ); + } + /** * Automatically stops the consumer when the garbage collector kicks in. */ @@ -116,7 +188,7 @@ abstract class AbstractConsumer implements Iterator * * @return bool Returns false when there are no pending messages. */ - public function stop($drop = false) + public function stop(bool $drop = false): bool { if (!$this->valid()) { return false; @@ -138,67 +210,18 @@ abstract class AbstractConsumer implements Iterator } /** - * Closes the underlying connection when forcing a disconnection. + * {@inheritdoc} */ - abstract protected function disconnect(); - - /** - * Writes a Redis command on the underlying connection. - * - * @param string $method Command ID. - * @param array $arguments Arguments for the command. - */ - abstract protected function writeRequest($method, $arguments); - - /** - * @return void - */ - #[ReturnTypeWillChange] - public function rewind() - { - // NOOP - } - - /** - * Returns the last message payload retrieved from the server and generated - * by one of the active subscriptions. - * - * @return array - */ - #[ReturnTypeWillChange] public function current() { return $this->getValue(); } /** - * @return int|null - */ - #[ReturnTypeWillChange] - public function key() - { - return $this->position; - } - - /** - * @return int|null - */ - #[ReturnTypeWillChange] - public function next() - { - if ($this->valid()) { - ++$this->position; - } - - return $this->position; - } - - /** - * Checks if the the consumer is still in a valid state to continue. + * Checks if the consumer is still in a valid state to continue. * * @return bool */ - #[ReturnTypeWillChange] public function valid() { $isValid = $this->isFlagSet(self::STATUS_VALID); @@ -217,10 +240,59 @@ abstract class AbstractConsumer implements Iterator } /** - * Waits for a new message from the server generated by one of the active - * subscriptions and returns it when available. - * - * @return array + * {@inheritdoc} */ - abstract protected function getValue(); + protected function disconnect() + { + $this->client->disconnect(); + } + + /** + * {@inheritdoc} + */ + protected function getValue() + { + /** @var NodeConnectionInterface $connection */ + $connection = $this->client->getConnection(); + $response = $connection->read(); + + switch ($response[0]) { + case self::SUBSCRIBE: + case self::UNSUBSCRIBE: + case self::PSUBSCRIBE: + case self::PUNSUBSCRIBE: + if ($response[2] === 0) { + $this->invalidate(); + } + // The missing break here is intentional as we must process + // subscriptions and unsubscriptions as standard messages. + // no break + + case self::MESSAGE: + return (object) [ + 'kind' => $response[0], + 'channel' => $response[1], + 'payload' => $response[2], + ]; + + case self::PMESSAGE: + return (object) [ + 'kind' => $response[0], + 'pattern' => $response[1], + 'channel' => $response[2], + 'payload' => $response[3], + ]; + + case self::PONG: + return (object) [ + 'kind' => $response[0], + 'payload' => $response[1], + ]; + + default: + throw new ClientException( + "Unknown message type '{$response[0]}' received in the PUB/SUB context." + ); + } + } } diff --git a/src/Consumer/PubSub/DispatcherLoop.php b/src/Consumer/PubSub/DispatcherLoop.php new file mode 100644 index 00000000..558b41ba --- /dev/null +++ b/src/Consumer/PubSub/DispatcherLoop.php @@ -0,0 +1,108 @@ +consumer = $consumer; + } + + /** + * Binds a callback to a channel. + * + * @param string $messageType Channel name. + * @param callable $callback A callback. + */ + public function attachCallback(string $messageType, callable $callback): void + { + $callbackName = $this->getPrefixKeys() . $messageType; + + $this->callbacksDictionary[$callbackName] = $callback; + $this->consumer->subscribe($messageType); + } + + /** + * Stops listening to a channel and removes the associated callback. + * + * @param string $messageType Redis channel. + */ + public function detachCallback(string $messageType): void + { + $callbackName = $this->getPrefixKeys() . $messageType; + + if (isset($this->callbacksDictionary[$callbackName])) { + unset($this->callbacksDictionary[$callbackName]); + $this->consumer->unsubscribe($messageType); + } + } + + /** + * Starts the dispatcher loop. + */ + public function run(): void + { + foreach ($this->consumer as $message) { + $kind = $message->kind; + + if ($kind !== Consumer::MESSAGE && $kind !== Consumer::PMESSAGE) { + if (isset($this->defaultCallback)) { + $callback = $this->defaultCallback; + $callback($message, $this); + } + + continue; + } + + if (isset($this->callbacksDictionary[$message->channel])) { + $callback = $this->callbacksDictionary[$message->channel]; + $callback($message->payload, $this); + } elseif (isset($this->defaultCallback)) { + $callback = $this->defaultCallback; + $callback($message, $this); + } + } + } + + /** + * Return the prefix used for keys. + * + * @return string + */ + protected function getPrefixKeys(): string + { + $options = $this->consumer->getClient()->getOptions(); + + if (isset($options->prefix)) { + /** @var KeyPrefixProcessor $processor */ + $processor = $options->prefix; + + return $processor->getPrefix(); + } + + return ''; + } +} diff --git a/src/PubSub/RelayConsumer.php b/src/Consumer/PubSub/RelayConsumer.php similarity index 94% rename from src/PubSub/RelayConsumer.php rename to src/Consumer/PubSub/RelayConsumer.php index 2af67b84..18b9c0d8 100644 --- a/src/PubSub/RelayConsumer.php +++ b/src/Consumer/PubSub/RelayConsumer.php @@ -10,7 +10,7 @@ * file that was distributed with this source code. */ -namespace Predis\PubSub; +namespace Predis\Consumer\PubSub; use Predis\NotSupportedException; @@ -99,16 +99,8 @@ class RelayConsumer extends Consumer /** * {@inheritDoc} */ - public function stop($drop = false) + public function stop($drop = false): bool { return false; } - - /** - * {@inheritDoc} - */ - public function __destruct() - { - // NOOP - } } diff --git a/src/Consumer/Push/Consumer.php b/src/Consumer/Push/Consumer.php new file mode 100644 index 00000000..f656c491 --- /dev/null +++ b/src/Consumer/Push/Consumer.php @@ -0,0 +1,55 @@ +client); + } + } + + /** + * @return PushResponseInterface|null + */ + public function current(): ?PushResponseInterface + { + return parent::current(); + } + + /** + * Reads line from connection and returns push response or null on any other type. + * + * @return PushResponseInterface|null + */ + protected function getValue(): ?PushResponseInterface + { + /** @var NodeConnectionInterface $connection */ + $connection = $this->client->getConnection(); + $response = $connection->read(); + + return ($response instanceof PushResponse) ? $response : null; + } +} diff --git a/src/Consumer/Push/DispatcherLoop.php b/src/Consumer/Push/DispatcherLoop.php new file mode 100644 index 00000000..3d57eb4b --- /dev/null +++ b/src/Consumer/Push/DispatcherLoop.php @@ -0,0 +1,43 @@ +consumer = $consumer; + } + + /** + * {@inheritDoc} + */ + public function run(): void + { + foreach ($this->consumer as $notification) { + if (null !== $notification) { + $messageType = $notification->getDataType(); + + if (isset($this->callbacksDictionary[$messageType])) { + $callback = $this->callbacksDictionary[$messageType]; + $callback($notification->getPayload(), $this); + } elseif (isset($this->defaultCallback)) { + $callback = $this->defaultCallback; + $callback($notification->getPayload(), $this); + } + } + } + } +} diff --git a/src/Consumer/Push/PushNotificationException.php b/src/Consumer/Push/PushNotificationException.php new file mode 100644 index 00000000..d3081b30 --- /dev/null +++ b/src/Consumer/Push/PushNotificationException.php @@ -0,0 +1,19 @@ +response = $serverResponse; + } + + /** + * {@inheritDoc} + * @throws PushNotificationException + */ + public function getDataType(): string + { + if (!isset($this->response[0])) { + throw new PushNotificationException('Invalid server response'); + } + + return $this->response[0]; + } + + /** + * {@inheritDoc} + */ + public function getPayload(): array + { + return array_slice($this->response, 1); + } + + public function offsetExists($offset): bool + { + return isset($this->response[$offset]); + } + + #[ReturnTypeWillChange] + public function offsetGet($offset) + { + return $this->response[$offset]; + } + + public function offsetSet($offset, $value): void + { + $this->response[$offset] = $value; + } + + public function offsetUnset($offset): void + { + unset($this->response[$offset]); + } +} diff --git a/src/Consumer/Push/PushResponseInterface.php b/src/Consumer/Push/PushResponseInterface.php new file mode 100644 index 00000000..7460d8ec --- /dev/null +++ b/src/Consumer/Push/PushResponseInterface.php @@ -0,0 +1,37 @@ +checkCapabilities($client); - - $this->options = $options ?: []; - $this->client = $client; - - $this->genericSubscribeInit('subscribe'); - $this->genericSubscribeInit('psubscribe'); - } - - /** - * Returns the underlying client instance used by the pub/sub iterator. - * - * @return ClientInterface - */ - public function getClient() - { - return $this->client; - } - - /** - * Checks if the client instance satisfies the required conditions needed to - * initialize a PUB/SUB consumer. - * - * @param ClientInterface $client Client instance used by the consumer. - * - * @throws NotSupportedException - */ - protected function checkCapabilities(ClientInterface $client) - { - if ($client->getConnection() instanceof ClusterInterface) { - throw new NotSupportedException( - 'Cannot initialize a PUB/SUB consumer over cluster connections.' - ); - } - - $commands = ['publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe']; - - if (!$client->getCommandFactory()->supports(...$commands)) { - throw new NotSupportedException( - 'PUB/SUB commands are not supported by the current command factory.' - ); - } - } - - /** - * This method shares the logic to handle both SUBSCRIBE and PSUBSCRIBE. - * - * @param string $subscribeAction Type of subscription. - */ - protected function genericSubscribeInit($subscribeAction) - { - if (isset($this->options[$subscribeAction])) { - $this->$subscribeAction($this->options[$subscribeAction]); - } - } - - /** - * {@inheritdoc} - */ - protected function writeRequest($method, $arguments) - { - $this->client->getConnection()->writeRequest( - $this->client->createCommand($method, - Command::normalizeArguments($arguments) - ) - ); - } - - /** - * {@inheritdoc} - */ - protected function disconnect() - { - $this->client->disconnect(); - } - - /** - * {@inheritdoc} - */ - protected function getValue() - { - $response = $this->client->getConnection()->read(); - - switch ($response[0]) { - case self::SUBSCRIBE: - case self::UNSUBSCRIBE: - case self::PSUBSCRIBE: - case self::PUNSUBSCRIBE: - if ($response[2] === 0) { - $this->invalidate(); - } - // The missing break here is intentional as we must process - // subscriptions and unsubscriptions as standard messages. - // no break - - case self::MESSAGE: - return (object) [ - 'kind' => $response[0], - 'channel' => $response[1], - 'payload' => $response[2], - ]; - - case self::PMESSAGE: - return (object) [ - 'kind' => $response[0], - 'pattern' => $response[1], - 'channel' => $response[2], - 'payload' => $response[3], - ]; - - case self::PONG: - return (object) [ - 'kind' => $response[0], - 'payload' => $response[1], - ]; - - default: - throw new ClientException( - "Unknown message type '{$response[0]}' received in the PUB/SUB context." - ); - } - } -} diff --git a/src/PubSub/DispatcherLoop.php b/src/PubSub/DispatcherLoop.php deleted file mode 100644 index b6f79cd3..00000000 --- a/src/PubSub/DispatcherLoop.php +++ /dev/null @@ -1,171 +0,0 @@ -callbacks = []; - $this->pubsub = $pubsub; - } - - /** - * Checks if the passed argument is a valid callback. - * - * @param mixed $callable A callback. - * - * @throws InvalidArgumentException - */ - protected function assertCallback($callable) - { - if (!is_callable($callable)) { - throw new InvalidArgumentException('The given argument must be a callable object.'); - } - } - - /** - * Returns the underlying PUB / SUB context. - * - * @return Consumer - */ - public function getPubSubConsumer() - { - return $this->pubsub; - } - - /** - * Sets a callback that gets invoked upon new subscriptions. - * - * @param mixed $callable A callback. - */ - public function subscriptionCallback($callable = null) - { - if (isset($callable)) { - $this->assertCallback($callable); - } - - $this->subscriptionCallback = $callable; - } - - /** - * Sets a callback that gets invoked when a message is received on a - * channel that does not have an associated callback. - * - * @param mixed $callable A callback. - */ - public function defaultCallback($callable = null) - { - if (isset($callable)) { - $this->assertCallback($callable); - } - - $this->subscriptionCallback = $callable; - } - - /** - * Binds a callback to a channel. - * - * @param string $channel Channel name. - * @param callable $callback A callback. - */ - public function attachCallback($channel, $callback) - { - $callbackName = $this->getPrefixKeys() . $channel; - - $this->assertCallback($callback); - $this->callbacks[$callbackName] = $callback; - $this->pubsub->subscribe($channel); - } - - /** - * Stops listening to a channel and removes the associated callback. - * - * @param string $channel Redis channel. - */ - public function detachCallback($channel) - { - $callbackName = $this->getPrefixKeys() . $channel; - - if (isset($this->callbacks[$callbackName])) { - unset($this->callbacks[$callbackName]); - $this->pubsub->unsubscribe($channel); - } - } - - /** - * Starts the dispatcher loop. - */ - public function run() - { - foreach ($this->pubsub as $message) { - $kind = $message->kind; - - if ($kind !== Consumer::MESSAGE && $kind !== Consumer::PMESSAGE) { - if (isset($this->subscriptionCallback)) { - $callback = $this->subscriptionCallback; - call_user_func($callback, $message, $this); - } - - continue; - } - - if (isset($this->callbacks[$message->channel])) { - $callback = $this->callbacks[$message->channel]; - call_user_func($callback, $message->payload, $this); - } elseif (isset($this->defaultCallback)) { - $callback = $this->defaultCallback; - call_user_func($callback, $message, $this); - } - } - } - - /** - * Terminates the dispatcher loop. - */ - public function stop() - { - $this->pubsub->stop(); - } - - /** - * Return the prefix used for keys. - * - * @return string - */ - protected function getPrefixKeys() - { - $options = $this->pubsub->getClient()->getOptions(); - - if (isset($options->prefix)) { - return $options->prefix->getPrefix(); - } - - return ''; - } -} diff --git a/tests/Predis/ClientTest.php b/tests/Predis/ClientTest.php index a8043538..3b9eefd8 100644 --- a/tests/Predis/ClientTest.php +++ b/tests/Predis/ClientTest.php @@ -1048,7 +1048,7 @@ class ClientTest extends PredisTestCase { $client = new Client(); - $this->assertInstanceOf('Predis\PubSub\Consumer', $client->pubSubLoop()); + $this->assertInstanceOf('Predis\Consumer\PubSub\Consumer', $client->pubSubLoop()); } /** @@ -1061,7 +1061,7 @@ class ClientTest extends PredisTestCase $client = new Client($connection); - $this->assertInstanceOf('Predis\PubSub\Consumer', $pubsub = $client->pubSubLoop($options)); + $this->assertInstanceOf('Predis\Consumer\PubSub\Consumer', $pubsub = $client->pubSubLoop($options)); $reflection = new ReflectionProperty($pubsub, 'options'); $reflection->setAccessible(true); @@ -1123,11 +1123,11 @@ class ClientTest extends PredisTestCase ->method('__invoke') ->withConsecutive( [ - $this->isInstanceOf('Predis\PubSub\Consumer'), + $this->isInstanceOf('Predis\Consumer\PubSub\Consumer'), (object) ['kind' => 'subscribe', 'channel' => 'channel', 'payload' => 1], ], [ - $this->isInstanceOf('Predis\PubSub\Consumer'), + $this->isInstanceOf('Predis\Consumer\PubSub\Consumer'), (object) ['kind' => 'unsubscribe', 'channel' => 'channel', 'payload' => 0], ] ) diff --git a/tests/Predis/Consumer/AbstractConsumerTest.php b/tests/Predis/Consumer/AbstractConsumerTest.php new file mode 100644 index 00000000..00fb66ef --- /dev/null +++ b/tests/Predis/Consumer/AbstractConsumerTest.php @@ -0,0 +1,134 @@ +mockClient = $this->getMockBuilder(ClientInterface::class)->getMock(); + + $this->testClass = new class($this->mockClient) extends AbstractConsumer { + protected function getValue() + { + return 'payload'; + } + }; + } + + /** + * @group disconnected + * @return void + */ + public function testStopWithConnectionDrop(): void + { + $this->mockClient + ->expects($this->once()) + ->method('disconnect') + ->withAnyParameters(); + + $this->assertTrue($this->testClass->stop(true)); + } + + /** + * @group disconnected + * @return void + */ + public function testStopWithoutConnectionDrop(): void + { + $this->mockClient + ->expects($this->never()) + ->method('disconnect') + ->withAnyParameters(); + + $this->assertTrue($this->testClass->stop()); + } + + /** + * @group disconnected + * @return void + */ + public function testGetClient(): void + { + $this->assertSame($this->mockClient, $this->testClass->getClient()); + } + + /** + * @group disconnected + * @return void + */ + public function testCurrentReturnsCurrentPayload(): void + { + $this->assertSame('payload', $this->testClass->current()); + } + + /** + * @group disconnected + * @return void + */ + public function testValidReturnConsumerState(): void + { + $this->assertTrue($this->testClass->valid()); + + $this->testClass->stop(); + + $this->assertFalse($this->testClass->valid()); + } + + /** + * @group disconnected + * @return void + */ + public function testKeyReturnsCurrentPosition(): void + { + $this->assertSame(0, $this->testClass->key()); + + $this->testClass->next(); + + $this->assertSame(1, $this->testClass->key()); + } + + /** + * @group disconnected + * @return void + */ + public function testNextIncrementPositionOnValidState(): void + { + $this->assertSame(0, $this->testClass->key()); + + $this->testClass->next(); + + $this->assertSame(1, $this->testClass->key()); + + $this->testClass->stop(); + $this->testClass->next(); + + $this->assertSame(1, $this->testClass->key()); + } +} diff --git a/tests/Predis/Consumer/AbstractDispatcherLoopTest.php b/tests/Predis/Consumer/AbstractDispatcherLoopTest.php new file mode 100644 index 00000000..6c2395c9 --- /dev/null +++ b/tests/Predis/Consumer/AbstractDispatcherLoopTest.php @@ -0,0 +1,130 @@ +getMockBuilder(ClientInterface::class)->getMock(); + $this->mockConsumer = $this + ->getMockBuilder(ConsumerInterface::class) + ->setConstructorArgs([$mockClient]) + ->getMock(); + + $this->testClass = new class($this->mockConsumer) extends AbstractDispatcherLoop { + public function run(): void + { + // NOOP + } + + public function getCallbacks(): array + { + return $this->callbacksDictionary; + } + + public function getDefaultCallback(): callable + { + return $this->defaultCallback; + } + }; + } + + /** + * @group disconnected + * @return void + */ + public function testGetConsumer(): void + { + $this->assertSame($this->mockConsumer, $this->testClass->getConsumer()); + } + + /** + * @group disconnected + * @return void + */ + public function testSetDefaultCallback(): void + { + $callback = static function () { + return 'test'; + }; + + $this->testClass->setDefaultCallback($callback); + + $this->assertSame($callback, $this->testClass->getDefaultCallback()); + } + + /** + * @group disconnected + * @return void + */ + public function testAttachCallback(): void + { + $callback = static function () { + return 'test'; + }; + + $this->testClass->attachCallback('type', $callback); + + $this->assertSame(['type' => $callback], $this->testClass->getCallbacks()); + } + + /** + * @group disconnected + * @return void + */ + public function testDetachCallback(): void + { + $callback = static function () { + return 'test'; + }; + + $this->testClass->attachCallback('type', $callback); + + $this->assertSame(['type' => $callback], $this->testClass->getCallbacks()); + + $this->testClass->detachCallback('type'); + + $this->assertSame([], $this->testClass->getCallbacks()); + } + + /** + * @group disconnected + * @return void + */ + public function testStop(): void + { + $this->mockConsumer + ->expects($this->once()) + ->method('stop') + ->withAnyParameters(); + + $this->testClass->stop(); + } +} diff --git a/tests/Predis/PubSub/ConsumerTest.php b/tests/Predis/Consumer/PubSub/ConsumerTest.php similarity index 99% rename from tests/Predis/PubSub/ConsumerTest.php rename to tests/Predis/Consumer/PubSub/ConsumerTest.php index 6c68877c..f419aea6 100644 --- a/tests/Predis/PubSub/ConsumerTest.php +++ b/tests/Predis/Consumer/PubSub/ConsumerTest.php @@ -10,10 +10,10 @@ * file that was distributed with this source code. */ -namespace Predis\PubSub; +namespace Predis\Consumer\PubSub; use Predis\Client; -use Predis\PubSub\Consumer as PubSubConsumer; +use Predis\Consumer\PubSub\Consumer as PubSubConsumer; use PredisTestCase; /** @@ -172,7 +172,6 @@ class ConsumerTest extends PredisTestCase $pubsub = new PubSubConsumer($client); $this->assertFalse($pubsub->valid()); - $this->assertNull($pubsub->next()); } /** diff --git a/tests/Predis/PubSub/DispatcherLoopTest.php b/tests/Predis/Consumer/PubSub/DispatcherLoopTest.php similarity index 99% rename from tests/Predis/PubSub/DispatcherLoopTest.php rename to tests/Predis/Consumer/PubSub/DispatcherLoopTest.php index d41b4cea..60c7e687 100644 --- a/tests/Predis/PubSub/DispatcherLoopTest.php +++ b/tests/Predis/Consumer/PubSub/DispatcherLoopTest.php @@ -10,7 +10,7 @@ * file that was distributed with this source code. */ -namespace Predis\PubSub; +namespace Predis\Consumer\PubSub; use Predis\Client; use PredisTestCase; diff --git a/tests/Predis/Consumer/Push/ConsumerTest.php b/tests/Predis/Consumer/Push/ConsumerTest.php new file mode 100644 index 00000000..2e1aba14 --- /dev/null +++ b/tests/Predis/Consumer/Push/ConsumerTest.php @@ -0,0 +1,88 @@ +mockClient = $this->getMockBuilder(ClientInterface::class)->getMock(); + } + + /** + * @dataProvider responseProvider + * @group disconnected + * @param $readData + * @param $expectedResponse + * @return void + */ + public function testCurrentReturnsResponseFromServer($readData, $expectedResponse): void + { + $mockConnection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock(); + $mockConnection + ->expects($this->once()) + ->method('read') + ->withAnyParameters() + ->willReturn($readData); + + $this->mockClient + ->expects($this->once()) + ->method('getConnection') + ->withAnyParameters() + ->willReturn($mockConnection); + + $consumer = new Consumer($this->mockClient); + + $this->assertSame($expectedResponse, $consumer->current()); + } + + /** + * @group disconnected + * @return void + */ + public function testConstructCallsGivenCallbackOnObjectInstantiation(): void + { + $this->mockClient + ->expects($this->once()) + ->method('disconnect') + ->withAnyParameters(); + + $callback = static function (ClientInterface $client) { + $client->disconnect(); + }; + + new Consumer($this->mockClient, $callback); + } + + public function responseProvider(): array + { + $pushResponse = new PushResponse(['messageType', 'payload']); + + return [ + 'with push response' => [$pushResponse, $pushResponse], + 'with another response' => ['string', null], + ]; + } +}