m2_g2_fs/usr/local/lib/python3.10/dist-packages/redis/multidb/event.py

from typing import List

from redis.client import Redis
from redis.event import EventListenerInterface, OnCommandsFailEvent
from redis.multidb.database import SyncDatabase
from redis.multidb.failure_detector import FailureDetector


class ActiveDatabaseChanged:
    """
    Event fired when an active database has been changed.
    """

    def __init__(
        self,
        old_database: SyncDatabase,
        new_database: SyncDatabase,
        command_executor,
        **kwargs,
    ):
        self._old_database = old_database
        self._new_database = new_database
        self._command_executor = command_executor
        self._kwargs = kwargs

    @property
    def old_database(self) -> SyncDatabase:
        return self._old_database

    @property
    def new_database(self) -> SyncDatabase:
        return self._new_database

    @property
    def command_executor(self):
        return self._command_executor

    @property
    def kwargs(self):
        return self._kwargs


class ResubscribeOnActiveDatabaseChanged(EventListenerInterface):
    """
    Re-subscribe the currently active pub / sub to a new active database.
    """

    def listen(self, event: ActiveDatabaseChanged):
        old_pubsub = event.command_executor.active_pubsub

        if old_pubsub is not None:
            # Re-assign old channels and patterns so they will be automatically subscribed on connection.
            new_pubsub = event.new_database.client.pubsub(**event.kwargs)
            new_pubsub.channels = old_pubsub.channels
            new_pubsub.patterns = old_pubsub.patterns
            new_pubsub.shard_channels = old_pubsub.shard_channels
            new_pubsub.on_connect(None)
            event.command_executor.active_pubsub = new_pubsub
            old_pubsub.close()


class CloseConnectionOnActiveDatabaseChanged(EventListenerInterface):
    """
    Close connection to the old active database.
    """

    def listen(self, event: ActiveDatabaseChanged):
        event.old_database.client.close()

        if isinstance(event.old_database.client, Redis):
            event.old_database.client.connection_pool.update_active_connections_for_reconnect()
            event.old_database.client.connection_pool.disconnect()
        else:
            for node in event.old_database.client.nodes_manager.nodes_cache.values():
                node.redis_connection.connection_pool.update_active_connections_for_reconnect()
                node.redis_connection.connection_pool.disconnect()


class RegisterCommandFailure(EventListenerInterface):
    """
    Event listener that registers command failures and passing it to the failure detectors.
    """

    def __init__(self, failure_detectors: List[FailureDetector]):
        self._failure_detectors = failure_detectors

    def listen(self, event: OnCommandsFailEvent) -> None:
        for failure_detector in self._failure_detectors:
            failure_detector.register_failure(event.exception, event.commands)