Repository: redis/redis-py
File: redis/cluster.py (Python)

Code before:
    def _get_node_pubsub(self, node):
        try:
            return self.node_pubsub_mapping[node.name]
        except KeyError:
            pubsub = node.redis_connection.pubsub(
                push_handler_func=self.push_handler_func
            )
            self.node_pubsub_mapping[node.name] = pubsub
            return pubsub

    def _sharded_message_generator(self):
        """
        Iterate through pubsubs until a complete cycle is done.
        """
        while True:
            pubsub = next(self._pubsubs_generator)

            # None marks end of cycle
            if pubsub is None:
                break

            message = pubsub.get_message()
            if message is not None:
                return message

        return None

    def _pubsubs_generator(self):
        """
        Generator that yields pubsubs in round-robin fashion.
        Yields None to mark cycle boundaries.
        """
        while True:
            current_nodes = list(self.node_pubsub_mapping.values())
            yield from current_nodes
            yield None  # Cycle marker

    def get_sharded_message(
        self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
    ):
        if target_node:
            message = self.node_pubsub_mapping[target_node.name].get_message(
                ignore_subscribe_messages=ignore_subscribe_messages, timeout=timeout
            )
        else:
            message = self._sharded_message_generator()
        if message is None:
            return None
        elif str_if_bytes(message["type"]) == "sunsubscribe":
            if message["channel"] in self.pending_unsubscribe_shard_channels:
                self.pending_unsubscribe_shard_channels.remove(message["channel"])

Review comment (bug):
The infinite loop with break on None sentinel could hang if no messages arrive. The original implementation had a bounded iteration using `range(len(self.node_pubsub_mapping))` which provided a natural timeout mechanism. Consider either restoring the bounded iteration with the snapshot approach, or adding explicit timeout handling to prevent potential hangs.

Code after:
    def _get_node_pubsub(self, node):
        try:
            return self.node_pubsub_mapping[node.name]
        except KeyError:
            pubsub = node.redis_connection.pubsub(
                push_handler_func=self.push_handler_func
            )
            self.node_pubsub_mapping[node.name] = pubsub
            return pubsub

    def _sharded_message_generator(self):
        for _ in range(len(self.node_pubsub_mapping)):
            pubsub = next(self._pubsubs_generator)
            message = pubsub.get_message()
            if message is not None:
                return message
        return None

    def _pubsubs_generator(self):
        while True:
            current_nodes = list(self.node_pubsub_mapping.values())
            yield from current_nodes

    def get_sharded_message(
        self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
    ):
        if target_node:
            message = self.node_pubsub_mapping[target_node.name].get_message(
                ignore_subscribe_messages=ignore_subscribe_messages, timeout=timeout
            )
        else:
            message = self._sharded_message_generator()
        if message is None:
            return None
        elif str_if_bytes(message["type"]) == "sunsubscribe":
            if message["channel"] in self.pending_unsubscribe_shard_channels:
                self.pending_unsubscribe_shard_channels.remove(message["channel"])
                self.shard_channels.pop(message["channel"], None)
                node = self.cluster.get_node_from_key(message["channel"])
                if self.node_pubsub_mapping[node.name].subscribed is False:
                    self.node_pubsub_mapping.pop(node.name)
        if not self.channels and not self.patterns and not self.shard_channels:
            # There are no subscriptions anymore, set subscribed_event flag
            # to false
            self.subscribed_event.clear()
        if self.ignore_subscribe_messages or ignore_subscribe_messages:
            return None
        return message

    def ssubscribe(self, *args, **kwargs):
        if args: