geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-09-14 12:27:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Push notifications support (#1316)

Browse Source 
* 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
This commit is contained in:
Vladyslav Vildanov
2023-06-29 16:18:04 +03:00
committed by GitHub
parent f239e989fe
commit 6a1a2c4734
27 changed files with 1309 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/dispatcher_loop.php
+1 -1
View File
@@ -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
examples/push_notifications.php
+49
View File
@@ -0,0 +1,49 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
use Predis\ClientInterface;
use Predis\Consumer\Push\PushResponseInterface;
require __DIR__ . '/shared.php';
// 1. Create client with RESP3 protocol specified. Push notifications allowed only in RESP3 mode.
$client = new Predis\Client($single_server + ['read_write_timeout' => 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";
}
}
examples/push_notifications_dispatcher.php
+68
View File
@@ -0,0 +1,68 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
use Predis\ClientInterface;
use Predis\Consumer\DispatcherLoopInterface;
use Predis\Consumer\Push\DispatcherLoop;
use Predis\Consumer\Push\PushResponseInterface;
require __DIR__ . '/shared.php';
// 1. Create client with RESP3 protocol specified. Push notifications allowed only in RESP3 mode.
$client = new Predis\Client($single_server + ['read_write_timeout' => 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";
src/Client.php
+14 -2
View File
@@ -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.
*
src/ClientContextInterface.php
+5 -1
View File
@@ -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()
src/ClientInterface.php
+5 -1
View File
@@ -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()
src/Connection/StreamConnection.php
+11 -1
View File
@@ -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
*/
src/Consumer/AbstractConsumer.php
+112
View File
@@ -0,0 +1,112 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
use Predis\ClientInterface;
use ReturnTypeWillChange;
abstract class AbstractConsumer implements ConsumerInterface
{
/**
* @var ClientInterface
*/
protected $client;
/**
* @var bool
*/
protected $isValid = true;
/**
* @var int
*/
protected $position = 0;
public function __construct(ClientInterface $client)
{
$this->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
}
}
src/Consumer/AbstractDispatcherLoop.php
+86
View File
@@ -0,0 +1,86 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
abstract class AbstractDispatcherLoop implements DispatcherLoopInterface
{
/**
* @var ConsumerInterface
*/
protected $consumer;
/**
* @var callable|null
*/
protected $defaultCallback;
/**
* @var callable[]
*/
protected $callbacksDictionary;
/**
* {@inheritDoc}
*/
public function __construct(ConsumerInterface $consumer)
{
$this->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();
}
}
src/Consumer/ConsumerInterface.php
+64
View File
@@ -0,0 +1,64 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
use Iterator;
use Predis\ClientInterface;
use ReturnTypeWillChange;
interface ConsumerInterface extends Iterator
{
/**
* @param ClientInterface $client
*/
public function __construct(ClientInterface $client);
/**
* Stops consumer loop, with optional client disconnection.
*
* @param bool $drop
* @return bool
*/
public function stop(bool $drop = false): bool;
/**
* Returns consumer client instance.
*
* @return ClientInterface
*/
public function getClient(): ClientInterface;
/**
* Returns last payload produced by server.
*
* @return mixed
*/
#[ReturnTypeWillChange]
public function current();
/**
* Keeps loop until consumer is in valid state.
*
* @return void
*/
#[ReturnTypeWillChange]
public function next();
/**
* Checks if consumer is in the valid state to continue.
*
* @return bool
*/
#[ReturnTypeWillChange]
public function valid();
}
src/Consumer/DispatcherLoopInterface.php
+65
View File
@@ -0,0 +1,65 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
/**
* Abstraction around consumer interface to invoke callbacks on received messages.
*/
interface DispatcherLoopInterface
{
/**
* Returns consumer interface instance.
*
* @return ConsumerInterface
*/
public function getConsumer(): ConsumerInterface;
/**
* Sets default callback that invokes if message type have no matching callback.
*
* @param callable|null $callback
* @return void
*/
public function setDefaultCallback(callable $callback = null): void;
/**
* Binds given message type to given callback.
*
* @param string $messageType
* @param callable $callback
* @return void
*/
public function attachCallback(string $messageType, callable $callback): void;
/**
* Removes connection between given message type and previously assigned callback.
*
* @param string $messageType
* @return void
*/
public function detachCallback(string $messageType): void;
/**
* Starts consumer loop.
*
* @return void
*/
public function run(): void;
/**
* Stops consumer loop.
*
* @return void
*/
public function stop(): void;
}
src/PubSub/AbstractConsumer.php → src/Consumer/PubSub/Consumer.php
+135 -63
View File
@@ -10,15 +10,20 @@
* file that was distributed with this source code.
*/
namespace Predis\PubSub;
namespace Predis\Consumer\PubSub;
use Iterator;
use ReturnTypeWillChange;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\Command;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Consumer\AbstractConsumer;
use Predis\NotSupportedException;
/**
* Base implementation of a PUB/SUB consumer abstraction based on PHP iterators.
* PUB/SUB consumer abstraction.
*/
abstract class AbstractConsumer implements Iterator
class Consumer extends AbstractConsumer
{
public const SUBSCRIBE = 'subscribe';
public const UNSUBSCRIBE = 'unsubscribe';
@@ -32,9 +37,76 @@ abstract class AbstractConsumer implements Iterator
public const STATUS_SUBSCRIBED = 2; // 0b0010
public const STATUS_PSUBSCRIBED = 4; // 0b0100
protected $position;
protected $statusFlags = self::STATUS_VALID;
protected $options;
/**
* @param ClientInterface $client Client instance used by the consumer.
* @param array|null $options Options for the consumer initialization.
* @throws NotSupportedException
*/
public function __construct(ClientInterface $client, array $options = null)
{
parent::__construct($client);
$this->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."
);
}
}
}
src/Consumer/PubSub/DispatcherLoop.php
+108
View File
@@ -0,0 +1,108 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\PubSub;
use Predis\Command\Processor\KeyPrefixProcessor;
use Predis\Consumer\AbstractDispatcherLoop;
/**
* Method-dispatcher loop built around the client-side abstraction of a Redis
* PUB / SUB context.
*/
class DispatcherLoop extends AbstractDispatcherLoop
{
/**
* @var Consumer
*/
protected $consumer;
public function __construct(Consumer $consumer)
{
$this->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 '';
}
}
src/PubSub/RelayConsumer.php → src/Consumer/PubSub/RelayConsumer.php
+2 -10
View File
@@ -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
}
}
src/Consumer/Push/Consumer.php
+55
View File
@@ -0,0 +1,55 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use Predis\ClientInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Consumer\AbstractConsumer;
class Consumer extends AbstractConsumer
{
/**
* @param ClientInterface $client
* @param callable|null $preLoopCallback Callback that should be called on client before enter a loop.
*/
public function __construct(ClientInterface $client, callable $preLoopCallback = null)
{
parent::__construct($client);
if (null !== $preLoopCallback) {
$preLoopCallback($this->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;
}
}
src/Consumer/Push/DispatcherLoop.php
+43
View File
@@ -0,0 +1,43 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use Predis\Consumer\AbstractDispatcherLoop;
class DispatcherLoop extends AbstractDispatcherLoop
{
public function __construct(Consumer $consumer)
{
$this->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);
}
}
}
}
}
src/Consumer/Push/PushNotificationException.php
+19
View File
@@ -0,0 +1,19 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use Exception;
class PushNotificationException extends Exception
{
}
src/Consumer/Push/PushResponse.php
+71
View File
@@ -0,0 +1,71 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use ArrayAccess;
use ReturnTypeWillChange;
class PushResponse implements PushResponseInterface, ArrayAccess
{
/**
* @var array
*/
private $response;
public function __construct(array $serverResponse)
{
$this->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]);
}
}
src/Consumer/Push/PushResponseInterface.php
+37
View File
@@ -0,0 +1,37 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use Predis\Response\ResponseInterface;
interface PushResponseInterface extends ResponseInterface
{
public const PUB_SUB_DATA_TYPE = 'pubsub';
public const MONITOR_DATA_TYPE = 'monitor';
public const INVALIDATE_DATA_TYPE = 'invalidate';
public const MESSAGE_DATA_TYPE = 'message';
/**
* Returns PUSH notification data type.
*
* @return string
*/
public function getDataType(): string;
/**
* Returns PUSH notification payload.
*
* @return array
*/
public function getPayload(): array;
}
src/PubSub/Consumer.php
-157
View File
@@ -1,157 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\PubSub;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\Command;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\NotSupportedException;
/**
* PUB/SUB consumer.
*/
class Consumer extends AbstractConsumer
{
protected $client;
protected $options;
/**
* @param ClientInterface $client Client instance used by the consumer.
* @param array $options Options for the consumer initialization.
*/
public function __construct(ClientInterface $client, array $options = null)
{
$this->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."
);
}
}
}
src/PubSub/DispatcherLoop.php
-171
View File
@@ -1,171 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\PubSub;
use InvalidArgumentException;
/**
* Method-dispatcher loop built around the client-side abstraction of a Redis
* PUB / SUB context.
*/
class DispatcherLoop
{
private $pubsub;
protected $callbacks;
protected $defaultCallback;
protected $subscriptionCallback;
/**
* @param Consumer $pubsub PubSub consumer instance used by the loop.
*/
public function __construct(Consumer $pubsub)
{
$this->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 '';
}
}
tests/Predis/ClientTest.php
+4 -4
View File
@@ -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],
]
)
tests/Predis/Consumer/AbstractConsumerTest.php
+134
View File
@@ -0,0 +1,134 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
use PHPUnit\Framework\MockObject\MockObject;
use Predis\ClientInterface;
use PredisTestCase;
class AbstractConsumerTest extends PredisTestCase
{
/**
* @var ConsumerInterface
*/
private $testClass;
/**
* @var MockObject&ClientInterface&MockObject
*/
private $mockClient;
protected function setUp(): void
{
parent::setUp();
$this->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());
}
}
tests/Predis/Consumer/AbstractDispatcherLoopTest.php
+130
View File
@@ -0,0 +1,130 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer;
use PHPUnit\Framework\MockObject\MockObject;
use Predis\ClientInterface;
use PredisTestCase;
class AbstractDispatcherLoopTest extends PredisTestCase
{
/**
* @var DispatcherLoopInterface
*/
private $testClass;
/**
* @var MockObject&ConsumerInterface&MockObject
*/
private $mockConsumer;
protected function setUp(): void
{
parent::setUp();
$mockClient = $this->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();
}
}
tests/Predis/PubSub/ConsumerTest.php → tests/Predis/Consumer/PubSub/ConsumerTest.php
+2 -3
View File
@@ -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());
}
/**
tests/Predis/PubSub/DispatcherLoopTest.php → tests/Predis/Consumer/PubSub/DispatcherLoopTest.php
+1 -1
View File
@@ -10,7 +10,7 @@
* file that was distributed with this source code.
*/
namespace Predis\PubSub;
namespace Predis\Consumer\PubSub;
use Predis\Client;
use PredisTestCase;
tests/Predis/Consumer/Push/ConsumerTest.php
+88
View File
@@ -0,0 +1,88 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Consumer\Push;
use PHPUnit\Framework\MockObject\MockObject;
use Predis\ClientInterface;
use Predis\Connection\NodeConnectionInterface;
use PredisTestCase;
class ConsumerTest extends PredisTestCase
{
/**
* @var MockObject&ClientInterface&MockObject
*/
private $mockClient;
protected function setUp(): void
{
parent::setUp();
$this->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],
];
}
}