Method get_sharded_message

redis/cluster.py:3000–3064

(
        self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
    )

            yield from current_nodes

    def get_sharded_message(
        self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
    ):
        if target_node:
            # Use .get(): migration-driven cleanup in the sunsubscribe branch
            # below and reset() both remove entries from node_pubsub_mapping,
            # so a caller polling with target_node may race the cleanup. Match
            # the async counterpart's None-handling rather than raising
            # KeyError. None pubsub falls through to "no message available".
            pubsub = self.node_pubsub_mapping.get(target_node.name)
            if pubsub is not None:
                # Don't pass ignore_subscribe_messages here - let get_sharded_message
                # handle the filtering after processing subscription state changes
                message = pubsub.get_message(
                    ignore_subscribe_messages=False, timeout=timeout
                )
            else:
                message = None
        else:
            pubsub, message = self._sharded_message_generator(timeout=timeout)
        if message is None:
            return None
        # Only sunsubscribe mutates cluster-level shard state; bypassing the
        # lock on the data-message hot path keeps smessage delivery from
        # competing with the reconciliation worker for _shard_state_lock.
        if str_if_bytes(message["type"]) == "sunsubscribe":
            # Serialize state mutation against reinitialize_shard_subscriptions
            # (worker thread). The blocking get_message above intentionally
            # runs outside the lock so reconciliation is not stalled by long
            # polls.
            with self._shard_state_lock:
                if message["channel"] in self.pending_unsubscribe_shard_channels:
                    # User-initiated sunsubscribe: drop from cluster-level tracking.
                    self.pending_unsubscribe_shard_channels.remove(message["channel"])
                    self.shard_channels.pop(message["channel"], None)
                    self._shard_channel_to_node.pop(message["channel"], None)
                # Drop the per-node pubsub that delivered the confirmation once
                # it no longer holds any shard subscriptions, regardless of
                # whether the sunsubscribe was user-initiated or driven by
                # slot-migration reconciliation (_migrate_shard_channel, which
                # intentionally does not add the channel to
                # pending_unsubscribe_shard_channels). This releases the
                # dedicated connection that would otherwise linger.
                # Identifying the receiving pubsub directly (rather than via
                # the cluster's current slot map) is required after slot
                # migration, where the channel's owner is no longer the node
                # that received our original SSUBSCRIBE.
                if pubsub is not None and not pubsub.subscribed:
                    name = self._find_node_name_for_pubsub(pubsub)
                    if name is not None:
                        try:
                            pubsub.reset()
                        except Exception:
                            pass
                        self.node_pubsub_mapping.pop(name, None)
                # Mirror PubSub.handle_message: the empty-check belongs in the
                # unsubscribe branch since that is the only path that can
                # reduce shard_channels here.