Add posibility to use RedisCluster as connection Class

rq/queue.py

@@ -9,6 +9,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Type, Union
 
 from redis import WatchError
+from redis.exceptions import RedisClusterException
 
 from .timeouts import BaseDeathPenalty, UnixSignalDeathPenalty
 
@@ -1371,7 +1372,11 @@ def dequeue_any(
         while True:
             queue_keys = [q.key for q in queues]
             if len(queue_keys) == 1 and get_version(connection) >= (6, 2, 0):
-                result = cls.lmove(connection, queue_keys[0], timeout)
+                try:
+                    result = cls.lmove(connection, queue_keys[0], timeout)
+                    # If you using redis Cluster you can only use lpop
+                except RedisClusterException:
+                    result = cls.lpop(queue_keys, timeout, connection=connection)
             else:
                 result = cls.lpop(queue_keys, timeout, connection=connection)
             if result is None:

rq/worker.py

@@ -427,11 +427,21 @@ def _set_connection(self, connection: 'Redis') -> 'Redis':
         Args:
             connection (Optional[Redis]): The Redis Connection.
         """
-        current_socket_timeout = connection.connection_pool.connection_kwargs.get("socket_timeout")
-        if current_socket_timeout is None:
-            timeout_config = {"socket_timeout": self.connection_timeout}
-            connection.connection_pool.connection_kwargs.update(timeout_config)
-        return connection
+        try:
+            current_socket_timeout = connection.connection_pool.connection_kwargs.get("socket_timeout")
+            if current_socket_timeout is None:
+                timeout_config = {"socket_timeout": self.connection_timeout}
+                connection.connection_pool.connection_kwargs.update(timeout_config)
+            return connection
+        # If you are using RedisCluster you needs to pars all cluster nodes.
+        except AttributeError:
+            nodes = connection.get_nodes()
+            for node in nodes:
+                current_socket_timeout = node.redis_connection.connection_pool.connection_kwargs.get("socket_timeout")
+                if current_socket_timeout is None:
+                    timeout_config = {"socket_timeout": self.connection_timeout}
+                    node.redis_connection.connection_pool.connection_kwargs.update(timeout_config)
+            return connection
 
     @property
     def dequeue_timeout(self) -> int:

tests/__init__.py

@@ -75,11 +75,9 @@ class RQTestCase(unittest.TestCase):
     def setUpClass(cls):
         # Set up connection to Redis
         testconn = find_empty_redis_database()
-
         # Store the connection (for sanity checking)
         cls.testconn = testconn
         cls.connection = testconn
-
         # Shut up logging
         logging.disable(logging.ERROR)