diff --git a/README.md b/README.md
index d7375470..61815878 100644
--- a/README.md
+++ b/README.md
@@ -159,6 +159,7 @@ when needed. The client options supported by default in Predis are:
   - `aggregate`: configures the client with a custom aggregate connection (callable).
   - `parameters`: list of default connection parameters for aggregate connections.
   - `commands`: specifies a command factory instance to use through the library.
+  - `readTimeout`: (cluster only) Timeout between read operations while loop over connections.
 
 Users can also provide custom options with values or callable objects (for lazy initialization) that
 are stored in the options container for later use through the library.
diff --git a/examples/sharded_dispatcher_loop.php b/examples/sharded_dispatcher_loop.php
new file mode 100644
index 00000000..31553636
--- /dev/null
+++ b/examples/sharded_dispatcher_loop.php
@@ -0,0 +1,91 @@
+<?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\Client;
+
+require __DIR__ . '/shared.php';
+
+// This is a basic example on how to use the Predis\DispatcherLoop class in sharded pub/sub mode.
+//
+// Both channels belong to different shards. So in sharded mode we can subscribe
+// and receive messages from different shard channels.
+//
+// To see this example in action you can just use redis-cli and publish some
+// messages to the '{channels}_events' and 'control' channel, e.g.:
+
+// ./redis-cli
+// SPUBLISH {channels}_events first
+// SPUBLISH {channels}_events second
+// SPUBLISH {channels}_events third
+// SPUBLISH control terminate_dispatcher
+
+// 1. Create client and setup RW timeout to 0.
+$client = new Client(
+    [
+        'tcp://127.0.0.1:6372?read_write_timeout=0',
+        'tcp://127.0.0.1:6373?read_write_timeout=0',
+        'tcp://127.0.0.1:6374?read_write_timeout=0',
+    ], [
+    'cluster' => 'redis',
+]);
+
+// 2. Run pub/sub loop.
+$pubSub = $client->pubSubLoop();
+
+// 3. Create a dispatcher loop instance and attach a bunch of callbacks.
+$dispatcher = new \Predis\Consumer\PubSub\DispatcherLoop($pubSub);
+
+// 4. Demonstrate how to use a callable class as a callback for the dispatcher loop.
+class EventsListener implements Countable
+{
+    private $events;
+
+    public function __construct()
+    {
+        $this->events = [];
+    }
+
+    public function count()
+    {
+        return count($this->events);
+    }
+
+    public function getEvents()
+    {
+        return $this->events;
+    }
+
+    public function __invoke($payload, $dispatcher)
+    {
+        $this->events[] = $payload;
+    }
+}
+
+// 5. Attach our callable class to the dispatcher.
+$dispatcher->attachCallback('{channels}_events', $events = new EventsListener());
+
+// 6. Attach a function to control the dispatcher loop termination with a message.
+$dispatcher->attachCallback('control', function ($payload, $dispatcher) {
+    if ($payload === 'terminate_dispatcher') {
+        $dispatcher->stop();
+    }
+});
+
+// 7. Run the dispatcher loop until the callback attached to the 'control' channel
+// receives 'terminate_dispatcher' as a message.
+$dispatcher->run();
+
+// Display our achievements!
+echo "We received {$events->count()} messages!", PHP_EOL;
+
+// Say goodbye :-)
+echo 'Goodbye from Redis!', PHP_EOL;
diff --git a/examples/sharded_pubsub_consumer.php b/examples/sharded_pubsub_consumer.php
new file mode 100644
index 00000000..a31a6454
--- /dev/null
+++ b/examples/sharded_pubsub_consumer.php
@@ -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.
+ */
+
+use Predis\Client;
+
+require __DIR__ . '/shared.php';
+
+// 1. Create client and setup RW timeout to 0.
+$client = new Client(
+    [
+        'tcp://127.0.0.1:6372?read_write_timeout=0',
+        'tcp://127.0.0.1:6373?read_write_timeout=0',
+        'tcp://127.0.0.1:6374?read_write_timeout=0',
+    ], [
+    'cluster' => 'redis',
+]);
+
+// 2. Run pub/sub loop. Sharded channels belongs to different shards.
+$pubSub = $client->pubSubLoop();
+$pubSub->ssubscribe('{channels}_notifications');
+$pubSub->ssubscribe('control_channel');
+
+// Start processing the pubsup messages. Open a terminal and use redis-cli
+// to push messages to the channels. Examples:
+//   ./redis-cli SPUBLISH {channels}_notifications "this is a test"
+//   ./redis-cli SPUBLISH control_channel quit_loop
+foreach ($pubSub as $message) {
+    switch ($message->kind) {
+        case 'ssubscribe':
+            echo "Subscribed to {$message->channel}", PHP_EOL;
+            break;
+
+        case 'message':
+            if ($message->channel == 'control_channel') {
+                if ($message->payload == 'quit_loop') {
+                    echo 'Aborting pubsub loop...', PHP_EOL;
+                    $pubSub->sunsubscribe();
+                } else {
+                    echo "Received an unrecognized command: {$message->payload}.", PHP_EOL;
+                }
+            } else {
+                echo "Received the following message from {$message->channel}:",
+                PHP_EOL, "  {$message->payload}", PHP_EOL, PHP_EOL;
+            }
+            break;
+    }
+}
+
+// Always unset the pubsub consumer instance when you are done! The
+// class destructor will take care of cleanups and prevent protocol
+// desynchronizations between the client and the server.
+unset($pubsub);
+
+// Say goodbye :-)
+echo 'Goodbye from Redis!', PHP_EOL;
diff --git a/src/ClientContextInterface.php b/src/ClientContextInterface.php
index d76a59ca..f2ed60b0 100644
--- a/src/ClientContextInterface.php
+++ b/src/ClientContextInterface.php
@@ -330,6 +330,9 @@ use Predis\Command\Redis\Container\Search\FTCURSOR;
  * @method $this save()
  * @method $this slaveof($host, $port)
  * @method $this slowlog($subcommand, $argument = null)
+ * @method $this spublish(string $shardChannel, string $message)
+ * @method $this ssubscribe(string ...$shardChannels)
+ * @method $this sunsubscribe(string ...$shardChannels)
  * @method $this time()
  * @method $this command()
  * @method $this geoadd($key, $longitude, $latitude, $member)
diff --git a/src/ClientInterface.php b/src/ClientInterface.php
index 143b1e9e..44176155 100644
--- a/src/ClientInterface.php
+++ b/src/ClientInterface.php
@@ -348,6 +348,9 @@ use Predis\Response\Status;
  * @method mixed             save()
  * @method mixed             slaveof(string $host, int $port)
  * @method mixed             slowlog($subcommand, $argument = null)
+ * @method int               spublish(string $shardChannel, string $message)
+ * @method array             ssubscribe(string ...$shardChannels)
+ * @method array             sunsubscribe(string ...$shardChannels)
  * @method array             time()
  * @method array             command()
  * @method int               geoadd(string $key, $longitude, $latitude, $member)
diff --git a/src/Cluster/ClusterStrategy.php b/src/Cluster/ClusterStrategy.php
index b12b8b0f..7ff365f3 100644
--- a/src/Cluster/ClusterStrategy.php
+++ b/src/Cluster/ClusterStrategy.php
@@ -170,6 +170,11 @@ abstract class ClusterStrategy implements StrategyInterface
             'GEODIST' => $getKeyFromFirstArgument,
             'GEORADIUS' => [$this, 'getKeyFromGeoradiusCommands'],
             'GEORADIUSBYMEMBER' => [$this, 'getKeyFromGeoradiusCommands'],
+
+            /* sharded pubsub */
+            'SSUBSCRIBE' => $getKeyFromAllArguments,
+            'SUNSUBSCRIBE' => [$this, 'getKeyFromSUnsubscribeCommand'],
+            'SPUBLISH' => $getKeyFromFirstArgument,
         ];
     }
 
@@ -388,6 +393,24 @@ abstract class ClusterStrategy implements StrategyInterface
         return $arguments[0];
     }
 
+    /**
+     * Extracts key from SUNSUBSCRIBE command if it's given.
+     *
+     * @param  CommandInterface $command
+     * @return string
+     */
+    protected function getKeyFromSUnsubscribeCommand(CommandInterface $command): ?string
+    {
+        $arguments = $command->getArguments();
+
+        // SUNSUBSCRIBE command could be called without arguments, so it doesn't matter on each node it will be called.
+        if (empty($arguments)) {
+            return 'fake';
+        }
+
+        return $this->getKeyFromAllArguments($command);
+    }
+
     /**
      * Extracts the key from EVAL and EVALSHA commands.
      *
diff --git a/src/Command/Redis/SPUBLISH.php b/src/Command/Redis/SPUBLISH.php
new file mode 100644
index 00000000..d6452189
--- /dev/null
+++ b/src/Command/Redis/SPUBLISH.php
@@ -0,0 +1,33 @@
+<?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\Command\Redis;
+
+use Predis\Command\PrefixableCommand as RedisCommand;
+
+/**
+ * @see https://redis.io/commands/spublish/
+ *
+ * Posts a message to the given shard channel.
+ */
+class SPUBLISH extends RedisCommand
+{
+    public function getId()
+    {
+        return 'SPUBLISH';
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForFirstArgument($prefix);
+    }
+}
diff --git a/src/Command/Redis/SSUBSCRIBE.php b/src/Command/Redis/SSUBSCRIBE.php
new file mode 100644
index 00000000..a923b3f4
--- /dev/null
+++ b/src/Command/Redis/SSUBSCRIBE.php
@@ -0,0 +1,33 @@
+<?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\Command\Redis;
+
+use Predis\Command\PrefixableCommand as RedisCommand;
+
+/**
+ * @see https://redis.io/commands/ssubscribe/
+ *
+ * Subscribes the client to the specified shard channels.
+ */
+class SSUBSCRIBE extends RedisCommand
+{
+    public function getId()
+    {
+        return 'SSUBSCRIBE';
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForAllArguments($prefix);
+    }
+}
diff --git a/src/Command/Redis/SUNSUBSCRIBE.php b/src/Command/Redis/SUNSUBSCRIBE.php
new file mode 100644
index 00000000..25caca4e
--- /dev/null
+++ b/src/Command/Redis/SUNSUBSCRIBE.php
@@ -0,0 +1,33 @@
+<?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\Command\Redis;
+
+use Predis\Command\PrefixableCommand as RedisCommand;
+
+/**
+ * @see https://redis.io/commands/sunsubscribe/
+ *
+ * Unsubscribes the client from the given shard channels, or from all of them if none is given.
+ */
+class SUNSUBSCRIBE extends RedisCommand
+{
+    public function getId()
+    {
+        return 'SUNSUBSCRIBE';
+    }
+
+    public function prefixKeys($prefix)
+    {
+        $this->applyPrefixForAllArguments($prefix);
+    }
+}
diff --git a/src/Configuration/Option/Cluster.php b/src/Configuration/Option/Cluster.php
index 34b33de4..9284df64 100644
--- a/src/Configuration/Option/Cluster.php
+++ b/src/Configuration/Option/Cluster.php
@@ -17,6 +17,7 @@ use Predis\Cluster\RedisStrategy;
 use Predis\Configuration\OptionsInterface;
 use Predis\Connection\Cluster\PredisCluster;
 use Predis\Connection\Cluster\RedisCluster;
+use Predis\Connection\Parameters;
 
 /**
  * Configures an aggregate connection used for clustering
@@ -58,8 +59,15 @@ class Cluster extends Aggregate
         switch ($description) {
             case 'redis':
             case 'redis-cluster':
-                return function ($parameters, $options, $option) {
-                    return new RedisCluster($options->connections, new RedisStrategy($options->crc16));
+                return static function ($parameters, $options, $option) {
+                    $optionParameters = $options->parameters ?? [];
+
+                    return new RedisCluster(
+                        $options->connections,
+                        new Parameters($optionParameters),
+                        new RedisStrategy($options->crc16),
+                        $options->readTimeout
+                    );
                 };
 
             case 'predis':
@@ -81,8 +89,10 @@ class Cluster extends Aggregate
      */
     protected function getDefaultConnectionInitializer()
     {
-        return function ($parameters, $options, $option) {
-            return new PredisCluster();
+        return static function ($parameters, $options, $option) {
+            $optionsParameters = $options->parameters ?? [];
+
+            return new PredisCluster(new Parameters($optionsParameters));
         };
     }
 
diff --git a/src/Configuration/OptionsInterface.php b/src/Configuration/OptionsInterface.php
index 597a0579..f627e35e 100644
--- a/src/Configuration/OptionsInterface.php
+++ b/src/Configuration/OptionsInterface.php
@@ -22,6 +22,7 @@ use Predis\Command\Processor\ProcessorInterface;
  * @property ProcessorInterface                  $prefix      Key prefixing strategy using the supplied string as prefix
  * @property \Predis\Command\FactoryInterface    $commands    Command factory for creating Redis commands
  * @property callable                            $replication Aggregate connection initializer for replication
+ * @property int                                 $readTimeout Timeout in milliseconds between read operations on reading from multiple connections.
  */
 interface OptionsInterface
 {
diff --git a/src/Connection/AbstractConnection.php b/src/Connection/AbstractConnection.php
index 6a2f7e5a..e8be6c62 100644
--- a/src/Connection/AbstractConnection.php
+++ b/src/Connection/AbstractConnection.php
@@ -79,6 +79,14 @@ abstract class AbstractConnection implements NodeConnectionInterface
         return isset($this->resource);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function hasDataToRead(): bool
+    {
+        return true;
+    }
+
     /**
      * {@inheritdoc}
      */
diff --git a/src/Connection/Cluster/PredisCluster.php b/src/Connection/Cluster/PredisCluster.php
index cacf0553..6af90367 100644
--- a/src/Connection/Cluster/PredisCluster.php
+++ b/src/Connection/Cluster/PredisCluster.php
@@ -56,10 +56,12 @@ class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
     private $connectionParameters;
 
     /**
-     * @param StrategyInterface $strategy Optional cluster strategy.
+     * @param ParametersInterface    $parameters
+     * @param StrategyInterface|null $strategy   Optional cluster strategy.
      */
-    public function __construct(StrategyInterface $strategy = null)
+    public function __construct(ParametersInterface $parameters, StrategyInterface $strategy = null)
     {
+        $this->connectionParameters = $parameters;
         $this->strategy = $strategy ?: new PredisStrategy();
         $this->distributor = $this->strategy->getDistributor();
     }
@@ -105,10 +107,6 @@ class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
     {
         $parameters = $connection->getParameters();
 
-        if (!isset($this->connectionParameters)) {
-            $this->connectionParameters = $parameters;
-        }
-
         $this->pool[(string) $connection] = $connection;
 
         if (isset($parameters->alias)) {
@@ -131,10 +129,6 @@ class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
                 unset($this->aliases[$alias]);
             }
 
-            if (empty($this->pool) && isset($this->connectionParameters)) {
-                $this->connectionParameters = null;
-            }
-
             return true;
         }
 
@@ -259,7 +253,7 @@ class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
     /**
      * {@inheritdoc}
      */
-    public function getParameters(): ?ParametersInterface
+    public function getParameters(): ParametersInterface
     {
         return $this->connectionParameters;
     }
diff --git a/src/Connection/Cluster/RedisCluster.php b/src/Connection/Cluster/RedisCluster.php
index 078e1d91..8a07d244 100644
--- a/src/Connection/Cluster/RedisCluster.php
+++ b/src/Connection/Cluster/RedisCluster.php
@@ -57,6 +57,10 @@ use Traversable;
 class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
 {
     private $useClusterSlots = true;
+
+    /**
+     * @var NodeConnectionInterface[]
+     */
     private $pool = [];
     private $slots = [];
     private $slotmap;
@@ -65,22 +69,35 @@ class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
     private $retryLimit = 5;
     private $retryInterval = 10;
 
+    /**
+     * @var int
+     */
+    private $readTimeout = 1000;
+
     /**
      * @var ParametersInterface
      */
     private $connectionParameters;
 
     /**
-     * @param FactoryInterface  $connections Optional connection factory.
-     * @param StrategyInterface $strategy    Optional cluster strategy.
+     * @param FactoryInterface       $connections Optional connection factory.
+     * @param StrategyInterface|null $strategy    Optional cluster strategy.
+     * @param int|null               $readTimeout Optional read timeout
      */
     public function __construct(
         FactoryInterface $connections,
-        StrategyInterface $strategy = null
+        ParametersInterface $parameters,
+        StrategyInterface $strategy = null,
+        int $readTimeout = null
     ) {
         $this->connections = $connections;
+        $this->connectionParameters = $parameters;
         $this->strategy = $strategy ?: new RedisClusterStrategy();
         $this->slotmap = new SlotMap();
+
+        if (!is_null($readTimeout)) {
+            $this->readTimeout = $readTimeout;
+        }
     }
 
     /**
@@ -156,10 +173,6 @@ class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
      */
     public function add(NodeConnectionInterface $connection)
     {
-        if (!isset($this->connectionParameters)) {
-            $this->connectionParameters = $connection->getParameters();
-        }
-
         $this->pool[(string) $connection] = $connection;
         $this->slotmap->reset();
     }
@@ -174,10 +187,6 @@ class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
             $this->slots = array_diff($this->slots, [$connection]);
             unset($this->pool[$id]);
 
-            if (empty($this->pool) && isset($this->connectionParameters)) {
-                $this->connectionParameters = null;
-            }
-
             return true;
         }
 
@@ -692,4 +701,22 @@ class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
     {
         return $this->connectionParameters;
     }
+
+    /**
+     * Loop over connections until there's data to read.
+     *
+     * @return mixed
+     */
+    public function read()
+    {
+        while (true) {
+            foreach ($this->pool as $connection) {
+                if ($connection->hasDataToRead()) {
+                    return $connection->read();
+                }
+            }
+
+            usleep($this->readTimeout);
+        }
+    }
 }
diff --git a/src/Connection/NodeConnectionInterface.php b/src/Connection/NodeConnectionInterface.php
index 71333177..2efbe1a5 100644
--- a/src/Connection/NodeConnectionInterface.php
+++ b/src/Connection/NodeConnectionInterface.php
@@ -54,4 +54,11 @@ interface NodeConnectionInterface extends ConnectionInterface
      * @return mixed
      */
     public function read();
+
+    /**
+     * Checks if current connection has data to read from server.
+     *
+     * @return bool
+     */
+    public function hasDataToRead(): bool;
 }
diff --git a/src/Connection/StreamConnection.php b/src/Connection/StreamConnection.php
index 8614de1f..2b546e3d 100644
--- a/src/Connection/StreamConnection.php
+++ b/src/Connection/StreamConnection.php
@@ -381,6 +381,25 @@ class StreamConnection extends AbstractConnection
         $this->write($buffer);
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    public function hasDataToRead(): bool
+    {
+        $resource = $this->getResource();
+
+        if ($resource) {
+            $resourceArray = [$resource];
+            $write = null;
+            $except = null;
+            $num = stream_select($resourceArray, $write, $except, 0);
+
+            return $num > 0;
+        }
+
+        return false;
+    }
+
     /**
      * Reads given resource split on chunks with given size.
      *
diff --git a/src/Consumer/PubSub/Consumer.php b/src/Consumer/PubSub/Consumer.php
index 23a368ad..11935a1b 100644
--- a/src/Consumer/PubSub/Consumer.php
+++ b/src/Consumer/PubSub/Consumer.php
@@ -16,6 +16,7 @@ use Predis\ClientException;
 use Predis\ClientInterface;
 use Predis\Command\Command;
 use Predis\Connection\Cluster\ClusterInterface;
+use Predis\Connection\ConnectionInterface;
 use Predis\Connection\NodeConnectionInterface;
 use Predis\Consumer\AbstractConsumer;
 use Predis\NotSupportedException;
@@ -26,7 +27,9 @@ use Predis\NotSupportedException;
 class Consumer extends AbstractConsumer
 {
     public const SUBSCRIBE = 'subscribe';
+    public const SSUBSCRIBE = 'ssubscribe';
     public const UNSUBSCRIBE = 'unsubscribe';
+    public const SUNSUBSCRIBE = 'sunsubscribe';
     public const PSUBSCRIBE = 'psubscribe';
     public const PUNSUBSCRIBE = 'punsubscribe';
     public const MESSAGE = 'message';
@@ -36,11 +39,17 @@ class Consumer extends AbstractConsumer
     public const STATUS_VALID = 1;       // 0b0001
     public const STATUS_SUBSCRIBED = 2;  // 0b0010
     public const STATUS_PSUBSCRIBED = 4; // 0b0100
+    public const STATUS_SSUBSCRIBED = 8; // 0b1000
 
     protected $statusFlags = self::STATUS_VALID;
 
     protected $options;
 
+    /**
+     * @var SubscriptionContext
+     */
+    private $subscriptionContext;
+
     /**
      * @param  ClientInterface       $client  Client instance used by the consumer.
      * @param  array|null            $options Options for the consumer initialization.
@@ -48,16 +57,29 @@ class Consumer extends AbstractConsumer
      */
     public function __construct(ClientInterface $client, array $options = null)
     {
+        $this->options = $options ?: [];
+        $this->setSubscriptionContext($client->getConnection());
+
         parent::__construct($client);
         $this->checkCapabilities($client);
 
-        $this->options = $options ?: [];
         $this->client = $client;
 
         $this->genericSubscribeInit('subscribe');
+        $this->genericSubscribeInit('ssubscribe');
         $this->genericSubscribeInit('psubscribe');
     }
 
+    /**
+     * Returns subscription context for current instance.
+     *
+     * @return SubscriptionContext
+     */
+    public function getSubscriptionContext(): SubscriptionContext
+    {
+        return $this->subscriptionContext;
+    }
+
     /**
      * Checks if the client instance satisfies the required conditions needed to
      * initialize a PUB/SUB consumer.
@@ -68,13 +90,7 @@ class Consumer extends AbstractConsumer
      */
     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'];
+        $commands = ['publish', 'spublish', 'subscribe', 'ssubscribe', 'unsubscribe', 'sunsubscribe', 'psubscribe', 'punsubscribe'];
 
         if (!$client->getCommandFactory()->supports(...$commands)) {
             throw new NotSupportedException(
@@ -84,7 +100,7 @@ class Consumer extends AbstractConsumer
     }
 
     /**
-     * This method shares the logic to handle both SUBSCRIBE and PSUBSCRIBE.
+     * This method shares the logic to handle SUBSCRIBE, SSUBSCRIBE, PSUBSCRIBE.
      *
      * @param string $subscribeAction Type of subscription.
      */
@@ -138,6 +154,17 @@ class Consumer extends AbstractConsumer
         $this->statusFlags |= self::STATUS_SUBSCRIBED;
     }
 
+    /**
+     * Subscribes to the specified shard channels.
+     *
+     * @param string ...$channels
+     */
+    public function ssubscribe(string ...$channels)
+    {
+        $this->writeRequest(self::SSUBSCRIBE, func_get_args());
+        $this->statusFlags |= self::STATUS_SSUBSCRIBED;
+    }
+
     /**
      * Unsubscribes from the specified channels.
      *
@@ -148,6 +175,16 @@ class Consumer extends AbstractConsumer
         $this->writeRequest(self::UNSUBSCRIBE, func_get_args());
     }
 
+    /**
+     * Unsubscribes from the specified shard channels.
+     *
+     * @param string ...$channels
+     */
+    public function sunsubscribe(string ...$channels)
+    {
+        $this->writeRequest(self::SUNSUBSCRIBE, func_get_args());
+    }
+
     /**
      * Subscribes to the specified channels using a pattern.
      *
@@ -204,6 +241,9 @@ class Consumer extends AbstractConsumer
             if ($this->isFlagSet(self::STATUS_PSUBSCRIBED)) {
                 $this->punsubscribe();
             }
+            if ($this->isFlagSet(self::STATUS_SSUBSCRIBED)) {
+                $this->sunsubscribe();
+            }
         }
 
         return !$drop;
@@ -225,7 +265,7 @@ class Consumer extends AbstractConsumer
     public function valid()
     {
         $isValid = $this->isFlagSet(self::STATUS_VALID);
-        $subscriptionFlags = self::STATUS_SUBSCRIBED | self::STATUS_PSUBSCRIBED;
+        $subscriptionFlags = self::STATUS_SUBSCRIBED | self::STATUS_PSUBSCRIBED | self::STATUS_SSUBSCRIBED;
         $hasSubscriptions = ($this->statusFlags & $subscriptionFlags) > 0;
 
         return $isValid && $hasSubscriptions;
@@ -258,7 +298,9 @@ class Consumer extends AbstractConsumer
 
         switch ($response[0]) {
             case self::SUBSCRIBE:
+            case self::SSUBSCRIBE:
             case self::UNSUBSCRIBE:
+            case self::SUNSUBSCRIBE:
             case self::PSUBSCRIBE:
             case self::PUNSUBSCRIBE:
                 if ($response[2] === 0) {
@@ -295,4 +337,19 @@ class Consumer extends AbstractConsumer
                 );
         }
     }
+
+    /**
+     * Set subscription context depends on connection.
+     *
+     * @param  NodeConnectionInterface $connection
+     * @return void
+     */
+    private function setSubscriptionContext(ConnectionInterface $connection): void
+    {
+        if ($connection instanceof ClusterInterface) {
+            $this->subscriptionContext = new SubscriptionContext(SubscriptionContext::CONTEXT_SHARDED);
+        } else {
+            $this->subscriptionContext = new SubscriptionContext();
+        }
+    }
 }
diff --git a/src/Consumer/PubSub/DispatcherLoop.php b/src/Consumer/PubSub/DispatcherLoop.php
index 558b41ba..a670d97f 100644
--- a/src/Consumer/PubSub/DispatcherLoop.php
+++ b/src/Consumer/PubSub/DispatcherLoop.php
@@ -42,7 +42,12 @@ class DispatcherLoop extends AbstractDispatcherLoop
         $callbackName = $this->getPrefixKeys() . $messageType;
 
         $this->callbacksDictionary[$callbackName] = $callback;
-        $this->consumer->subscribe($messageType);
+
+        if ($this->consumer->getSubscriptionContext()->getContext() === SubscriptionContext::CONTEXT_SHARDED) {
+            $this->consumer->ssubscribe($messageType);
+        } else {
+            $this->consumer->subscribe($messageType);
+        }
     }
 
     /**
@@ -56,7 +61,12 @@ class DispatcherLoop extends AbstractDispatcherLoop
 
         if (isset($this->callbacksDictionary[$callbackName])) {
             unset($this->callbacksDictionary[$callbackName]);
-            $this->consumer->unsubscribe($messageType);
+
+            if ($this->consumer->getSubscriptionContext()->getContext() === SubscriptionContext::CONTEXT_SHARDED) {
+                $this->consumer->sunsubscribe($messageType);
+            } else {
+                $this->consumer->unsubscribe($messageType);
+            }
         }
     }
 
diff --git a/src/Consumer/PubSub/SubscriptionContext.php b/src/Consumer/PubSub/SubscriptionContext.php
new file mode 100644
index 00000000..b4546a64
--- /dev/null
+++ b/src/Consumer/PubSub/SubscriptionContext.php
@@ -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\PubSub;
+
+class SubscriptionContext
+{
+    public const CONTEXT_SHARDED = 'sharded';
+    public const CONTEXT_NON_SHARDED = 'non_sharded';
+
+    /**
+     * @var string
+     */
+    private $context;
+
+    public function __construct(string $context = self::CONTEXT_NON_SHARDED)
+    {
+        $this->context = $context;
+    }
+
+    /**
+     * @return string
+     */
+    public function getContext(): string
+    {
+        return $this->context;
+    }
+}
diff --git a/tests/Predis/ClientTest.php b/tests/Predis/ClientTest.php
index 3b9eefd8..11e5e6c7 100644
--- a/tests/Predis/ClientTest.php
+++ b/tests/Predis/ClientTest.php
@@ -209,7 +209,7 @@ class ClientTest extends PredisTestCase
      */
     public function testConstructorWithClusterArgument(): void
     {
-        $cluster = new Connection\Cluster\PredisCluster();
+        $cluster = new Connection\Cluster\PredisCluster(new Parameters());
 
         $factory = new Connection\Factory();
         $cluster->add($factory->create('tcp://localhost:7000'));
@@ -1264,7 +1264,7 @@ class ClientTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6382');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6383');
 
-        $aggregate = new \Predis\Connection\Cluster\PredisCluster();
+        $aggregate = new \Predis\Connection\Cluster\PredisCluster(new Parameters());
 
         $aggregate->add($connection1);
         $aggregate->add($connection2);
diff --git a/tests/Predis/Cluster/PredisStrategyTest.php b/tests/Predis/Cluster/PredisStrategyTest.php
index fbf7922c..ed63ab51 100644
--- a/tests/Predis/Cluster/PredisStrategyTest.php
+++ b/tests/Predis/Cluster/PredisStrategyTest.php
@@ -216,6 +216,21 @@ class PredisStrategyTest extends PredisTestCase
         }
     }
 
+    /**
+     * @group disconnected
+     */
+    public function testKeysForSUnsubscribeCommand(): void
+    {
+        $strategy = $this->getClusterStrategy();
+        $commands = $this->getCommandFactory();
+        $arguments = [];
+
+        foreach ($this->getExpectedCommands('keys-sunsubscribe') as $commandID) {
+            $command = $commands->create($commandID, $arguments);
+            $this->assertNotNull($strategy->getSlot($command), $commandID);
+        }
+    }
+
     /**
      * @group disconnected
      */
@@ -444,6 +459,11 @@ class PredisStrategyTest extends PredisTestCase
             'GEODIST' => 'keys-first',
             'GEORADIUS' => 'keys-georadius',
             'GEORADIUSBYMEMBER' => 'keys-georadius',
+
+            /* sharded pubsub */
+            'SSUBSCRIBE' => 'keys-all',
+            'SUNSUBSCRIBE' => 'keys-sunsubscribe',
+            'SPUBLISH' => 'keys-first',
         ];
 
         if (isset($type)) {
diff --git a/tests/Predis/Cluster/RedisStrategyTest.php b/tests/Predis/Cluster/RedisStrategyTest.php
index dd570d53..b67b8cc4 100644
--- a/tests/Predis/Cluster/RedisStrategyTest.php
+++ b/tests/Predis/Cluster/RedisStrategyTest.php
@@ -231,6 +231,21 @@ class RedisStrategyTest extends PredisTestCase
         }
     }
 
+    /**
+     * @group disconnected
+     */
+    public function testKeysForSUnsubscribeCommand(): void
+    {
+        $strategy = $this->getClusterStrategy();
+        $commands = $this->getCommandFactory();
+        $arguments = [];
+
+        foreach ($this->getExpectedCommands('keys-sunsubscribe') as $commandID) {
+            $command = $commands->create($commandID, $arguments);
+            $this->assertNotNull($strategy->getSlot($command), $commandID);
+        }
+    }
+
     /**
      * @group disconnected
      */
@@ -467,6 +482,11 @@ class RedisStrategyTest extends PredisTestCase
             'GEODIST' => 'keys-first',
             'GEORADIUS' => 'keys-georadius',
             'GEORADIUSBYMEMBER' => 'keys-georadius',
+
+            /* sharded pubsub */
+            'SSUBSCRIBE' => 'keys-all',
+            'SUNSUBSCRIBE' => 'keys-sunsubscribe',
+            'SPUBLISH' => 'keys-first',
         ];
 
         if (isset($type)) {
diff --git a/tests/Predis/Command/Redis/SPUBLISH_Test.php b/tests/Predis/Command/Redis/SPUBLISH_Test.php
new file mode 100644
index 00000000..89b14bdd
--- /dev/null
+++ b/tests/Predis/Command/Redis/SPUBLISH_Test.php
@@ -0,0 +1,72 @@
+<?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\Command\Redis;
+
+use Predis\Command\Redis;
+
+class SPUBLISH_Test extends Redis\PredisCommandTestCase
+{
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return SPUBLISH::class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'SPUBLISH';
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArguments(): void
+    {
+        $arguments = ['channel', 'message'];
+        $expected = ['channel', 'message'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $this->assertSame(1, $this->getCommand()->parseResponse(1));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testPublishesMessagesToChannel(): void
+    {
+        $redis1 = $this->getClient();
+        $redis2 = $this->getClient();
+
+        $redis1->ssubscribe('channel:foo');
+
+        $this->assertSame(1, $redis2->spublish('channel:foo', 'bar'));
+        $this->assertSame(0, $redis2->spublish('channel:hoge', 'piyo'));
+    }
+}
diff --git a/tests/Predis/Command/Redis/SSUBSCRIBE_Test.php b/tests/Predis/Command/Redis/SSUBSCRIBE_Test.php
new file mode 100644
index 00000000..27f33962
--- /dev/null
+++ b/tests/Predis/Command/Redis/SSUBSCRIBE_Test.php
@@ -0,0 +1,102 @@
+<?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\Command\Redis;
+
+use Predis\Response\ServerException;
+
+class SSUBSCRIBE_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return SSUBSCRIBE::class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'SSUBSCRIBE';
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArguments(): void
+    {
+        $arguments = ['channel:foo', 'channel:bar'];
+        $expected = ['channel:foo', 'channel:bar'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $raw = ['ssubscribe', 'channel', 1];
+        $expected = ['ssubscribe', 'channel', 1];
+
+        $command = $this->getCommand();
+
+        $this->assertSame($expected, $command->parseResponse($raw));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testSubscribesToGivenShardedChannels(): void
+    {
+        $redis = $this->getClient();
+
+        $this->assertSame(['ssubscribe', 'channel1', 1], $redis->ssubscribe('channel1'));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testAllowsSUnsubscribeAfterSSubscribe(): void
+    {
+        $redis = $this->getClient();
+
+        $this->assertSame(['ssubscribe', 'channel1', 1], $redis->ssubscribe('channel1'));
+        $this->assertSame(['sunsubscribe', 'channel1', 0], $redis->sunsubscribe('channel1'));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testCannotSendOtherCommandsAfterSSubscribe(): void
+    {
+        $this->expectException(ServerException::class);
+        $this->expectExceptionMessageMatches('/ERR.*only .* allowed in this context/');
+
+        $redis = $this->getClient();
+
+        $redis->ssubscribe('channel:foo');
+        $redis->set('foo', 'bar');
+    }
+}
diff --git a/tests/Predis/Command/Redis/SUNSUBSCRIBE_Test.php b/tests/Predis/Command/Redis/SUNSUBSCRIBE_Test.php
new file mode 100644
index 00000000..ddf0d147
--- /dev/null
+++ b/tests/Predis/Command/Redis/SUNSUBSCRIBE_Test.php
@@ -0,0 +1,90 @@
+<?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\Command\Redis;
+
+class SUNSUBSCRIBE_Test extends PredisCommandTestCase
+{
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedCommand(): string
+    {
+        return SUNSUBSCRIBE::class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    protected function getExpectedId(): string
+    {
+        return 'SUNSUBSCRIBE';
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testFilterArguments(): void
+    {
+        $arguments = ['channel:foo', 'channel:bar'];
+        $expected = ['channel:foo', 'channel:bar'];
+
+        $command = $this->getCommand();
+        $command->setArguments($arguments);
+
+        $this->assertSame($expected, $command->getArguments());
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testParseResponse(): void
+    {
+        $raw = ['sunsubscribe', 'channel', 1];
+        $expected = ['sunsubscribe', 'channel', 1];
+
+        $command = $this->getCommand();
+
+        $this->assertSame($expected, $command->parseResponse($raw));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testUnsubscribesFromGivenShardedChannels(): void
+    {
+        $redis = $this->getClient();
+
+        $this->assertSame(['sunsubscribe', 'channel1', 0], $redis->sunsubscribe('channel1'));
+    }
+
+    /**
+     * @group connected
+     * @group relay-incompatible
+     * @requiresRedisVersion >= 7.0.0
+     */
+    public function testUnsubscribesFromAllSubscribedChannels(): void
+    {
+        $redis = $this->getClient();
+
+        $this->assertSame(['ssubscribe', 'channel:foo', 1], $redis->ssubscribe('channel:foo'));
+        $this->assertSame(['ssubscribe', 'channel:bar', 2], $redis->ssubscribe('channel:bar'));
+
+        [$_, $unsubscribed1, $_] = $redis->sunsubscribe();
+        [$_, $unsubscribed2, $_] = $redis->getConnection()->read();
+        $this->assertSameValues(['channel:foo', 'channel:bar'], [$unsubscribed1, $unsubscribed2]);
+
+        $this->assertSame('echoed', $redis->echo('echoed'));
+    }
+}
diff --git a/tests/Predis/Configuration/Option/ClusterTest.php b/tests/Predis/Configuration/Option/ClusterTest.php
index c1fdcbf9..7efcab2b 100644
--- a/tests/Predis/Configuration/Option/ClusterTest.php
+++ b/tests/Predis/Configuration/Option/ClusterTest.php
@@ -265,10 +265,6 @@ class ClusterTest extends PredisTestCase
 
         /** @var OptionsInterface|MockObject */
         $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock();
-        $options
-            ->expects($this->never())
-            ->method('__get')
-            ->with('connections');
 
         $this->assertInstanceOf('closure', $initializer = $option->filter($options, 'predis'));
         $this->assertInstanceOf('Predis\Connection\Cluster\PredisCluster', $initializer($parameters = []));
@@ -285,15 +281,17 @@ class ClusterTest extends PredisTestCase
         $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock();
 
         $options
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(3))
             ->method('__get')
             ->withConsecutive(
                 ['connections'],
-                ['crc16']
+                ['crc16'],
+                ['readTimeout']
             )
             ->willReturnOnConsecutiveCalls(
                 $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(),
-                $this->getMockBuilder('Predis\Cluster\Hash\HashGeneratorInterface')->getMock()
+                $this->getMockBuilder('Predis\Cluster\Hash\HashGeneratorInterface')->getMock(),
+                1000
             );
 
         $this->assertInstanceOf('closure', $initializer = $option->filter($options, 'redis'));
@@ -311,15 +309,17 @@ class ClusterTest extends PredisTestCase
         $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock();
 
         $options
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(3))
             ->method('__get')
             ->withConsecutive(
                 ['connections'],
-                ['crc16']
+                ['crc16'],
+                ['readTimeout']
             )
             ->willReturnOnConsecutiveCalls(
                 $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(),
-                $this->getMockBuilder('Predis\Cluster\Hash\HashGeneratorInterface')->getMock()
+                $this->getMockBuilder('Predis\Cluster\Hash\HashGeneratorInterface')->getMock(),
+                1000
             );
 
         $this->assertInstanceOf('closure', $initializer = $option->filter($options, 'redis-cluster'));
diff --git a/tests/Predis/Connection/Cluster/PredisClusterTest.php b/tests/Predis/Connection/Cluster/PredisClusterTest.php
index af8829b0..b800f584 100644
--- a/tests/Predis/Connection/Cluster/PredisClusterTest.php
+++ b/tests/Predis/Connection/Cluster/PredisClusterTest.php
@@ -13,7 +13,6 @@
 namespace Predis\Connection\Cluster;
 
 use Predis\Connection\Parameters;
-use Predis\Connection\ParametersInterface;
 use PredisTestCase;
 
 class PredisClusterTest extends PredisTestCase
@@ -23,7 +22,7 @@ class PredisClusterTest extends PredisTestCase
      */
     public function testExposesCommandHashStrategy(): void
     {
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
         $this->assertInstanceOf('Predis\Cluster\PredisStrategy', $cluster->getClusterStrategy());
     }
 
@@ -35,7 +34,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -53,7 +52,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001?alias=node01');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002?alias=node02');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -72,7 +71,7 @@ class PredisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:7003');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -85,24 +84,6 @@ class PredisClusterTest extends PredisTestCase
         $this->assertCount(1, $cluster);
     }
 
-    /**
-     * @group disconnected
-     */
-    public function testRemoveConnectionsUnsetParametersOnEmptyConnectionPool(): void
-    {
-        $connection = $this->getMockConnection('tcp://127.0.0.1:7001?alias=node01');
-
-        $cluster = new PredisCluster();
-
-        $cluster->add($connection);
-
-        $this->assertInstanceOf(ParametersInterface::class, $cluster->getParameters());
-
-        $cluster->remove($connection);
-
-        $this->assertNull($cluster->getParameters());
-    }
-
     /**
      * @group disconnected
      */
@@ -118,7 +99,7 @@ class PredisClusterTest extends PredisTestCase
             ->expects($this->once())
             ->method('connect');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -141,7 +122,7 @@ class PredisClusterTest extends PredisTestCase
             ->expects($this->once())
             ->method('disconnect');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -166,7 +147,7 @@ class PredisClusterTest extends PredisTestCase
             ->method('isConnected')
             ->willReturn(true);
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -191,7 +172,7 @@ class PredisClusterTest extends PredisTestCase
             ->method('isConnected')
             ->willReturn(false);
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -207,7 +188,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -229,7 +210,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -246,7 +227,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -267,7 +248,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -303,7 +284,7 @@ class PredisClusterTest extends PredisTestCase
 
         $ping = $this->getCommandFactory()->create('ping');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($this->getMockConnection('tcp://127.0.0.1:6379'));
 
@@ -320,7 +301,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -354,7 +335,7 @@ class PredisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('writeRequest');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -380,7 +361,7 @@ class PredisClusterTest extends PredisTestCase
             ->method('readResponse')
             ->with($command);
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -406,7 +387,7 @@ class PredisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('executeCommand');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -422,7 +403,7 @@ class PredisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:7001?alias=first');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:7002?alias=second');
 
-        $cluster = new PredisCluster();
+        $cluster = new PredisCluster(new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -447,8 +428,7 @@ class PredisClusterTest extends PredisTestCase
             'protocol' => '3',
         ]);
 
-        $cluster = new PredisCluster();
-        $cluster->add($connection);
+        $cluster = new PredisCluster($expectedParameters);
 
         $this->assertEquals($expectedParameters, $cluster->getParameters());
     }
diff --git a/tests/Predis/Connection/Cluster/RedisClusterTest.php b/tests/Predis/Connection/Cluster/RedisClusterTest.php
index 5c58c5f1..a8599674 100644
--- a/tests/Predis/Connection/Cluster/RedisClusterTest.php
+++ b/tests/Predis/Connection/Cluster/RedisClusterTest.php
@@ -17,7 +17,6 @@ use Predis\Cluster;
 use Predis\Command;
 use Predis\Connection;
 use Predis\Connection\Parameters;
-use Predis\Connection\ParametersInterface;
 use Predis\Response;
 use PredisTestCase;
 
@@ -30,7 +29,7 @@ class RedisClusterTest extends PredisTestCase
     {
         /** @var Connection\FactoryInterface */
         $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
 
         $this->assertSame($factory, $cluster->getConnectionFactory());
     }
@@ -40,7 +39,7 @@ class RedisClusterTest extends PredisTestCase
      */
     public function testUsesRedisClusterStrategyByDefault(): void
     {
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $this->assertInstanceOf('Predis\Cluster\RedisStrategy', $cluster->getClusterStrategy());
     }
@@ -53,7 +52,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Cluster\StrategyInterface */
         $strategy = $this->getMockBuilder('Predis\Cluster\StrategyInterface')->getMock();
 
-        $cluster = new RedisCluster(new Connection\Factory(), $strategy);
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters(), $strategy);
 
         $this->assertSame($strategy, $cluster->getClusterStrategy());
     }
@@ -66,7 +65,7 @@ class RedisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -85,7 +84,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6371');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -95,25 +94,6 @@ class RedisClusterTest extends PredisTestCase
         $this->assertCount(1, $cluster);
     }
 
-    /**
-     * @group disconnected
-     */
-    public function testRemoveConnectionsUnsetParametersOnEmptyConnectionPool(): void
-    {
-        $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
-        $connection = $this->getMockConnection('tcp://127.0.0.1:7001?alias=node01');
-
-        $cluster = new RedisCluster($factory);
-
-        $cluster->add($connection);
-
-        $this->assertInstanceOf(ParametersInterface::class, $cluster->getParameters());
-
-        $cluster->remove($connection);
-
-        $this->assertNull($cluster->getParameters());
-    }
-
     /**
      * @group disconnected
      */
@@ -122,7 +102,7 @@ class RedisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -141,7 +121,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -190,7 +170,7 @@ class RedisClusterTest extends PredisTestCase
                 return $connect2;
             });
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -223,7 +203,7 @@ class RedisClusterTest extends PredisTestCase
             ->expects($this->once())
             ->method('disconnect');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -248,7 +228,7 @@ class RedisClusterTest extends PredisTestCase
             ->method('isConnected')
             ->willReturn(true);
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -273,7 +253,7 @@ class RedisClusterTest extends PredisTestCase
             ->method('isConnected')
             ->willReturn(false);
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -291,7 +271,7 @@ class RedisClusterTest extends PredisTestCase
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6383?slots=10923-16383');
         $connection4 = $this->getMockConnection('tcp://127.0.0.1:6384');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -359,7 +339,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\Cluster\RedisCluster|MockObject */
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
             ->onlyMethods(['getRandomConnection'])
-            ->setConstructorArgs([$factory])
+            ->setConstructorArgs([$factory, new Parameters()])
             ->getMock();
         $cluster
             ->expects($this->once())
@@ -388,7 +368,7 @@ class RedisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
 
@@ -410,7 +390,7 @@ class RedisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -440,7 +420,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380?slots=5461-10922');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381?slots=10923-16383');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -470,7 +450,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380?slots=5461-5499,5600-10922');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381?slots=10923-10999,11001-16383');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -504,7 +484,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -529,7 +509,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -569,7 +549,7 @@ class RedisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('writeRequest');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -596,7 +576,7 @@ class RedisClusterTest extends PredisTestCase
             ->method('readResponse')
             ->with($command);
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -669,7 +649,7 @@ class RedisClusterTest extends PredisTestCase
             ])
             ->willReturn($connection4);
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -739,7 +719,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\Cluster\RedisCluster|MockObject */
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
             ->onlyMethods(['getRandomConnection'])
-            ->setConstructorArgs([$factory])
+            ->setConstructorArgs([$factory, new Parameters()])
             ->getMock();
         $cluster
             ->expects($this->never())
@@ -771,7 +751,7 @@ class RedisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('create');
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
 
         $cluster->executeCommand(
             Command\RawCommand::create('get', 'node:1001')
@@ -789,7 +769,7 @@ class RedisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('create');
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->askSlotMap();
 
         $this->assertCount(0, $cluster->getSlotMap());
@@ -846,7 +826,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\Cluster\RedisCluster|MockObject */
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
             ->onlyMethods(['getRandomConnection'])
-            ->setConstructorArgs([$factory])
+            ->setConstructorArgs([$factory, new Parameters()])
             ->getMock();
         $cluster
             ->expects($this->exactly(3))
@@ -912,7 +892,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\Cluster\RedisCluster|MockObject */
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
             ->onlyMethods(['getRandomConnection'])
-            ->setConstructorArgs([$factory])
+            ->setConstructorArgs([$factory, new Parameters()])
             ->getMock();
         $cluster
             ->expects($this->exactly(2))
@@ -938,7 +918,7 @@ class RedisClusterTest extends PredisTestCase
         $connection1 = $this->getMockConnection('tcp://127.0.0.1:6379');
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -989,7 +969,7 @@ class RedisClusterTest extends PredisTestCase
             ->expects($this->never())
             ->method('create');
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -1045,7 +1025,7 @@ class RedisClusterTest extends PredisTestCase
             ])
             ->willReturn($connection3);
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -1083,7 +1063,7 @@ class RedisClusterTest extends PredisTestCase
         $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
         $factory->expects($this->never())->method('create');
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -1133,7 +1113,7 @@ class RedisClusterTest extends PredisTestCase
             ])
             ->willReturn($connection3);
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -1178,7 +1158,7 @@ class RedisClusterTest extends PredisTestCase
             ])
             ->willReturn($connection2);
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
         $cluster->useClusterSlots(false);
 
         $cluster->add($connection1);
@@ -1222,7 +1202,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\FactoryInterface */
         $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
 
         $cluster->add($connection1);
 
@@ -1274,7 +1254,7 @@ class RedisClusterTest extends PredisTestCase
             ])
             ->willReturn($connection2);
 
-        $cluster = new RedisCluster($factory);
+        $cluster = new RedisCluster($factory, new Parameters());
 
         $cluster->add($connection1);
 
@@ -1292,7 +1272,7 @@ class RedisClusterTest extends PredisTestCase
 
         $ping = $this->getCommandFactory()->create('ping');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($this->getMockConnection('tcp://127.0.0.1:6379'));
 
@@ -1309,7 +1289,7 @@ class RedisClusterTest extends PredisTestCase
         $connection2 = $this->getMockConnection('tcp://127.0.0.1:6380?slots=5461-10922');
         $connection3 = $this->getMockConnection('tcp://127.0.0.1:6381?slots=10923-16383');
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
 
         $cluster->add($connection1);
         $cluster->add($connection2);
@@ -1342,7 +1322,7 @@ class RedisClusterTest extends PredisTestCase
                         $clusterDownError,
                         'foobar'));
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
         $cluster->useClusterSlots(false);
         $cluster->setRetryLimit(2);
         $cluster->add($connection1);
@@ -1374,7 +1354,7 @@ class RedisClusterTest extends PredisTestCase
                         $clusterDownError
                     ));
 
-        $cluster = new RedisCluster(new Connection\Factory());
+        $cluster = new RedisCluster(new Connection\Factory(), new Parameters());
         $cluster->useClusterSlots(false);
         $cluster->setRetryLimit(2);
         $cluster->add($connection1);
@@ -1435,7 +1415,7 @@ class RedisClusterTest extends PredisTestCase
         /** @var Connection\Cluster\RedisCluster|MockObject */
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
             ->onlyMethods(['getRandomConnection'])
-            ->setConstructorArgs([$factory])
+            ->setConstructorArgs([$factory, new Parameters()])
             ->getMock();
         $cluster
             ->expects($this->exactly(3))
@@ -1476,8 +1456,7 @@ class RedisClusterTest extends PredisTestCase
             'protocol' => '3',
         ]);
 
-        $cluster = new RedisCluster($factory);
-        $cluster->add($connection);
+        $cluster = new RedisCluster($factory, $expectedParameters);
 
         $this->assertEquals($expectedParameters, $cluster->getParameters());
     }
diff --git a/tests/Predis/Consumer/PubSub/ConsumerTest.php b/tests/Predis/Consumer/PubSub/ConsumerTest.php
index f419aea6..728b5897 100644
--- a/tests/Predis/Consumer/PubSub/ConsumerTest.php
+++ b/tests/Predis/Consumer/PubSub/ConsumerTest.php
@@ -13,7 +13,10 @@
 namespace Predis\Consumer\PubSub;
 
 use Predis\Client;
+use Predis\Connection\Cluster\ClusterInterface;
+use Predis\Connection\NodeConnectionInterface;
 use Predis\Consumer\PubSub\Consumer as PubSubConsumer;
+use Predis\NotSupportedException;
 use PredisTestCase;
 
 /**
@@ -43,15 +46,14 @@ class ConsumerTest extends PredisTestCase
     /**
      * @group disconnected
      */
-    public function testPubSubConsumerDoesNotWorkOnClusters(): void
+    public function testPubSubConsumerAllowsClusterConnectionOnShardedContext(): void
     {
-        $this->expectException('Predis\NotSupportedException');
-        $this->expectExceptionMessage('Cannot initialize a PUB/SUB consumer over cluster connections');
-
         $cluster = $this->getMockBuilder('Predis\Connection\Cluster\ClusterInterface')->getMock();
         $client = new Client($cluster);
 
         new PubSubConsumer($client);
+
+        $this->assertTrue(true);
     }
 
     /**
@@ -81,7 +83,7 @@ class ConsumerTest extends PredisTestCase
         $commands = $this->getCommandFactory();
 
         $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
-        $connection->expects($this->exactly(2))->method('writeRequest');
+        $connection->expects($this->exactly(3))->method('writeRequest');
 
         /** @var Client */
         $client = $this->getMockBuilder('Predis\Client')
@@ -90,14 +92,18 @@ class ConsumerTest extends PredisTestCase
             ->setConstructorArgs([$connection])
             ->getMock();
         $client
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(3))
             ->method('createCommand')
-            ->with($this->logicalOr($this->equalTo('subscribe'), $this->equalTo('psubscribe')))
+            ->with($this->logicalOr(
+                $this->equalTo('subscribe'),
+                $this->equalTo('psubscribe'),
+                $this->equalTo('ssubscribe')
+            ))
             ->willReturnCallback(function ($id, $args) use ($commands) {
                 return $commands->create($id, $args);
             });
 
-        $options = ['subscribe' => 'channel:foo', 'psubscribe' => 'channels:*'];
+        $options = ['subscribe' => 'channel:foo', 'ssubscribe' => 'channel:bar', 'psubscribe' => 'channels:*'];
 
         new PubSubConsumer($client, $options);
     }
@@ -133,6 +139,7 @@ class ConsumerTest extends PredisTestCase
         $commands = $this->getCommandFactory();
         $classUnsubscribe = $commands->getCommandClass('unsubscribe');
         $classPunsubscribe = $commands->getCommandClass('punsubscribe');
+        $classSunsubscribe = $commands->getCommandClass('sunsubscribe');
 
         $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
 
@@ -142,15 +149,16 @@ class ConsumerTest extends PredisTestCase
             ->setConstructorArgs([$connection])
             ->getMock();
 
-        $options = ['subscribe' => 'channel:foo', 'psubscribe' => 'channels:*'];
+        $options = ['subscribe' => 'channel:foo', 'ssubscribe' => 'channel:bar', 'psubscribe' => 'channels:*'];
         $pubsub = new PubSubConsumer($client, $options);
 
         $connection
-            ->expects($this->exactly(2))
+            ->expects($this->exactly(3))
             ->method('writeRequest')
             ->with($this->logicalOr(
                 $this->isInstanceOf($classUnsubscribe),
-                $this->isInstanceOf($classPunsubscribe)
+                $this->isInstanceOf($classPunsubscribe),
+                $this->isInstanceOf($classSunsubscribe)
             ));
 
         $pubsub->stop(false);
@@ -283,6 +291,28 @@ class ConsumerTest extends PredisTestCase
         $this->assertSame(1, $message->payload);
     }
 
+    /**
+     * @group disconnected
+     */
+    public function testReadsSSubscriptionMessageFromConnection(): void
+    {
+        $rawmessage = ['ssubscribe', 'channel:foo', 1];
+
+        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
+        $connection
+            ->expects($this->once())
+            ->method('read')
+            ->willReturn($rawmessage);
+
+        $client = new Client($connection);
+        $pubsub = new PubSubConsumer($client, ['ssubscribe' => 'channel:foo']);
+
+        $message = $pubsub->current();
+        $this->assertSame('ssubscribe', $message->kind);
+        $this->assertSame('channel:foo', $message->channel);
+        $this->assertSame(1, $message->payload);
+    }
+
     /**
      * @group disconnected
      */
@@ -305,6 +335,28 @@ class ConsumerTest extends PredisTestCase
         $this->assertSame(1, $message->payload);
     }
 
+    /**
+     * @group disconnected
+     */
+    public function testReadsSUnsubscriptionMessageFromConnection(): void
+    {
+        $rawmessage = ['sunsubscribe', 'channel:foo', 1];
+
+        $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
+        $connection
+            ->expects($this->once())
+            ->method('read')
+            ->willReturn($rawmessage);
+
+        $client = new Client($connection);
+        $pubsub = new PubSubConsumer($client, ['ssubscribe' => 'channel:foo']);
+
+        $message = $pubsub->current();
+        $this->assertSame('sunsubscribe', $message->kind);
+        $this->assertSame('channel:foo', $message->channel);
+        $this->assertSame(1, $message->payload);
+    }
+
     /**
      * @group disconnected
      */
@@ -344,6 +396,32 @@ class ConsumerTest extends PredisTestCase
         $this->assertSame($client, $pubsub->getClient());
     }
 
+    /**
+     * @dataProvider connectionsProvider
+     * @group disconnected
+     * @param  string                $connection
+     * @param  string                $context
+     * @return void
+     * @throws NotSupportedException
+     */
+    public function testGetSubscriptionContext(string $connection, string $context): void
+    {
+        $connection = $this->getMockBuilder($connection)->getMock();
+
+        $client = new Client($connection);
+        $pubsub = new PubSubConsumer($client);
+
+        $this->assertSame($context, $pubsub->getSubscriptionContext()->getContext());
+    }
+
+    public function connectionsProvider(): array
+    {
+        return [
+            [ClusterInterface::class, SubscriptionContext::CONTEXT_SHARDED],
+            [NodeConnectionInterface::class, SubscriptionContext::CONTEXT_NON_SHARDED],
+        ];
+    }
+
     // ******************************************************************** //
     // ---- INTEGRATION TESTS --------------------------------------------- //
     // ******************************************************************** //