[12.x] transaction for PredisClusterConnection and pipelines for PhpRedisClusterConnection

Author: vadimonus

src/Illuminate/Cache/RedisTagSet.php

@@ -79,18 +79,11 @@ public function entries()
      */
     public function flushStaleEntries()
     {
-        $flushStaleEntries = function ($pipe) {
+        $this->store->connection()->pipeline(function ($pipe) {
             foreach ($this->tagIds() as $tagKey) {
                 $pipe->zremrangebyscore($this->store->getPrefix().$tagKey, 0, Carbon::now()->getTimestamp());
             }
-        };
-
-        $connection = $this->store->connection();
-        if ($connection instanceof PhpRedisConnection) {
-            $flushStaleEntries($connection);
-        } else {
-            $this->store->connection()->pipeline($flushStaleEntries);
-        }
+        });
     }
 
     /**

src/Illuminate/Queue/RedisQueue.php

@@ -183,13 +183,7 @@ public function bulk($jobs, $data = '', $queue = null)
             }
         };
 
-        if ($connection instanceof PhpRedisClusterConnection) {
-            $connection->transaction($bulk);
-        } elseif ($connection instanceof PredisClusterConnection) {
-            $connection->pipeline($bulk);
-        } else {
-            $connection->pipeline(fn () => $connection->transaction($bulk));
-        }
+        $connection->pipeline(fn () => $connection->transaction($bulk));
     }
 
     /**

src/Illuminate/Redis/Connections/PhpRedisClusterConnection.php

@@ -81,6 +81,18 @@ public function flushAll()
         }
     }
 
+    /**
+     * Execute commands in a pipeline.
+     *
+     * @param  callable|null  $callback
+     * @return mixed
+     */
+    public function pipeline(?callable $callback = null)
+    {
+        // phphredis does not support pipelines for Redis Cluster
+        return tap($this, $callback);
+    }
+
     /**
      * Return default node to use for cluster.
      *

src/Illuminate/Redis/Connections/PredisClusterConnection.php

@@ -2,6 +2,7 @@
 
 namespace Illuminate\Redis\Connections;
 
+use Illuminate\Support\Arr;
 use Predis\Command\Redis\FLUSHALL;
 use Predis\Command\Redis\FLUSHDB;
 use Predis\Command\ServerFlushDatabase;
@@ -49,4 +50,16 @@ public function keys(string $pattern)
         }
         return array_merge(...$keys);
     }
+
+    /**
+     * Execute commands in a transaction.
+     *
+     * @param  callable|null  $callback
+     * @return mixed
+     */
+    public function transaction(?callable $callback = null)
+    {
+        // p does not support transacation for Redis Cluster
+        return tap($this, $callback);
+    }
 }