o e)d?e*d@e*fdAdBZ+dS)E MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc Cs||_|js |n|j|_|j|_|j|_ |j s!| n|j |_ |j dur.|n|j |_|j|j|j|_|j|_|j|_|jtgt|j |j|j|j|j|j|j|jd|_d|_t|_ t!|_"||_#d|_$g|_%d|_&dS)N)failure_detectors databases command_retryfailover_strategyfailover_attemptsfailover_delayevent_dispatcherauto_fallback_intervalF)'r& _databases health_checksdefault_health_checks_health_checkshealth_check_interval_health_check_intervalhealth_check_policyvalue_health_check_policyr%default_failure_detectors_failure_detectorsr(default_failover_strategy_failover_strategy set_databasesr,_auto_fallback_intervalr+_event_dispatcherr'_command_retryupdate_supported_errorsConnectionRefusedErrorr r)r*command_executor initializedasyncioLock_hc_lockr _bg_scheduler_config_recurring_hc_task _hc_tasks_half_open_state_task)selfr$rK[/root/parts/websockify/install/lib/python3.10/site-packages/redis/asyncio/multidb/client.py__init__)sL      zMultiDBClient.__init__rJreturncs|js |IdH|SN)rA initializerJrKrKrL __aenter__VszMultiDBClient.__aenter__csj|jr |j|jr|j|jD]}|q|jIdH|jjr3|jjj IdHdSdSrO) rGcancelrIrHr5closer@active_databaseclientaclose)rJhc_taskrKrKrLrW[s    zMultiDBClient.aclosec|IdHdSrOrWrJexc_type exc_value tracebackrKrKrL __aexit__kzMultiDBClient.__aexit__cs|IdHt|j|j|j|_d}|jD]\}}|j |j |j j t jkr4|s4||j_d}q|s;tdd|_dS)zT Perform initialization of databases to define their initial state. NFTz4Initial connection failed - no active database found)_perform_initial_health_checkrB create_taskrErun_recurring_asyncr2_check_databases_healthrGr-circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr@_active_databaserrA)rJis_active_db_founddatabaseweightrKrKrLrPns( zMultiDBClient.initializecCs|jS)zE Returns a sorted (by weight) list of all databases. )r-rQrKrKrL get_databasesszMultiDBClient.get_databasesrmNcsd}|jD] \}}||krd}nq|std||IdH|jjtjkr?|jdd\}}|j |t j IdHdSt d)zL Promote one of the existing databases to become an active. NT/Given database is not a member of database listrz1Cannot set active database, database is unhealthy) r- ValueError_check_db_healthrerhrirj get_top_nr@set_active_databaserMANUALr)rJrmexists existing_db_highest_weighted_dbrKrKrLrus& z!MultiDBClient.set_active_databaseTskip_initial_health_checkcs|jdtdtdi|jr|jjj|jfi|j}n"|jr7|jtdtd|jjj|jd}n |jjdi|j}|j durJ| n|j }t |||j |j d}z ||IdHWn tym|skYnw|jdd\}}|j||j |||IdHdS) z Adds a new database to the database list. Args: config: DatabaseConfig object that contains the database configuration. skip_initial_health_check: If True, adds the database even if it is unhealthy. retryr)retriesbackoff)connection_poolN)rVrernhealth_check_urlrqrK) client_kwargsupdaterrfrom_urlrF client_class from_pool set_retryredefault_circuit_breakerrrnrrsrr-rtadd_change_active_database)rJr$r{rVrermrzhighest_weightrKrKrL add_databasesD   zMultiDBClient.add_database new_databasehighest_weight_databasecs>|j|jkr|jjtjkr|j|tjIdHdSdSdSrO) rnrerhrirjr@rur AUTOMATIC)rJrrrKrKrLrs z%MultiDBClient._change_active_databasecsZ|j|}|jdd\}}||kr)|jjtjkr+|j|t j IdHdSdSdS)z< Removes a database from the database list. rqrN) r-removertrerhrirjr@rurrv)rJrmrnrzrrKrKrLremove_databases zMultiDBClient.remove_databaserncspd}|jD] \}}||krd}nq|std|jdd\}}|j||||_|||IdHdS)z< Updates a database from the database list. NTrprqr)r-rrrt update_weightrnr)rJrmrnrwrxryrzrrKrKrLupdate_database_weightsz$MultiDBClient.update_database_weightfailure_detectorcCs|j|dS)z> Adds a new failure detector to the database. N)r7append)rJrrKrKrLadd_failure_detectorsz"MultiDBClient.add_failure_detector healthcheckc sN|j4IdH|j|WdIdHdS1IdHs wYdS)z: Adds a new health check to the database. N)rDr0r)rJrrKrKrLadd_health_checks.zMultiDBClient.add_health_checkcs.|js |IdH|jj|i|IdHS)zB Executes a single command and return its result. N)rArPr@execute_commandrJargsoptionsrKrKrLrszMultiDBClient.execute_commandcCst|S)z: Enters into pipeline mode of the client. )PipelinerQrKrKrLpipeline'szMultiDBClient.pipelineF shard_hintvalue_from_callable watch_delayfuncrwatchesrrrcs:|js |IdH|jj|g|R|||dIdHS)z3 Executes callable as transaction. Nr)rArPr@execute_transaction)rJrrrrrrKrKrL transaction-s zMultiDBClient.transactioncs&|js |IdHt|fi|S)z Return a Publish/Subscribe object. With this object, you can subscribe to channels and listen for messages that get published to them. N)rArPPubSub)rJkwargsrKrKrLpubsubCszMultiDBClient.pubsubcsig|_|jD]\}}t||}||<|j|q tj|jddiIdH}fddt|j|D}|D]\}}t |t rZ|j }t j |j_tjd|jdd||<q=|S) zk Runs health checks as a recurring task. Runs health checks against all databases. return_exceptionsTNcsi|] \}}||qSrKrK).0taskresult task_to_dbrKrL ^sz9MultiDBClient._check_databases_health..z%Health check failed, due to exception)exc_infoF)rHr-rBrbrsrgatherzipitems isinstancerrmriOPENrerhloggerdebugoriginal_exception)rJrmryrresults db_resultsr unhealthy_dbrKrrLrdNs,    z%MultiDBClient._check_databases_healthcs|IdH}d}|jjtjkrd|v}n!|jjtjkr,t|t|dk}n |jjtj kr9d|v}|sDt d|jjdS)zj Runs initial health check and evaluate healthiness based on initial_health_check_policy. NTFz:Initial health check failed. Initial health check policy: ) rdrFinitial_health_check_policyr ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumlen ONE_AVAILABLEr)rJr is_healthyrKrKrLraps   z+MultiDBClient._perform_initial_health_checkcsX|j|j|IdH}|s|jjtjkrtj|j_|S|r*|jjtjkr*tj|j_|S)zO Runs health checks on the given database until first failure. N)r5executer0rerhrirrj)rJrmrrKrKrLrss   zMultiDBClient._check_db_healthre old_state new_statecCst}|tjkrt||j|_dS|tjkr0|tj kr0t d|jd| t t||tjkrF|tjkrHt d|jddSdSdS)Nz Database z- is unreachable. Failover has been initiated.z is reachable again.)rBget_running_loopri HALF_OPENrbrsrmrIrjrrwarning call_laterr _half_open_circuitinfo)rJrerrlooprKrKrLrgs   z/MultiDBClient._on_circuit_state_change_callback)rJr#rNr#)T),__name__ __module__ __qualname____doc__r rMrRrWr_rPrrorrur boolrrrfloatrrrrrrrrrrrr!rstrrrdictrrdrarsrrirgrKrKrKrLr#"sf -$ 1      "r#recCs tj|_dSrO)rirrh)rerKrKrLr rc@seZdZdZdefddZdddZd d Zd d Zd dZ de fddZ de fddZ dddZdddZd ddZddZdeefddZdS)!rzG Pipeline implementation for multiple logical Redis databases. rVcCsg|_||_dSrO)_command_stack_client)rJrVrKrKrLrMs zPipeline.__init__rJrNc|SrOrKrQrKrKrLrRzPipeline.__aenter__cs*|IdH|j|||IdHdSrO)resetrr_r[rKrKrLr_szPipeline.__aexit__cCs |SrO) _async_self __await__rQrKrKrLrrzPipeline.__await__crrOrKrQrKrKrLrrzPipeline._async_selfcCs t|jSrO)rrrQrKrKrL__len__s zPipeline.__len__cCsdS)z1Pipeline instances should always evaluate to TrueTrKrQrKrKrL__bool__szPipeline.__bool__Ncs g|_dSrO)rrQrKrKrLrs zPipeline.resetcs|IdHdS)zClose the pipelineN)rrQrKrKrLrWszPipeline.aclosecOs|j||f|S)ar Stage a command to be executed when execute() is next called Returns the current Pipeline object back so commands can be chained together, such as: pipe = pipe.set('foo', 'bar').incr('baz').decr('bang') At some other point, you can then run: pipe.execute(), which will execute all commands queued in the pipe. )rrrrKrKrLpipeline_execute_commands z!Pipeline.pipeline_execute_commandcOs|j|i|S)zAdds a command to the stack)rrJrrrKrKrLrszPipeline.execute_commandcsV|jjs |jIdHz|jjt|jIdHW|IdHS|IdHw)z0Execute all the commands in the current pipelineN)rrArPr@execute_pipelinetuplerrrQrKrKrLrs zPipeline.execute)rJrrNrrNN)rNr)rrrrr#rMrRr_rrintrrrrrWrrrrrrKrKrKrLrs    rc@seZdZdZdefddZd&ddZd'd d Zd d Ze de fd dZ de fddZ dedefddZdefddZdedefddZddZ d(de deefdd Zdd!d"d#eddfd$d%ZdS))rz2 PubSub object for multi database client. rVcKs||_|jjjdi|dS)zInitialize the PubSub object for a multi-database client. Args: client: MultiDBClient instance to use for pub/sub operations **kwargs: Additional keyword arguments to pass to the underlying pubsub implementation NrK)rr@r)rJrVrrKrKrLrMszPubSub.__init__rNcrrOrKrQrKrKrLrRrzPubSub.__aenter__NcrYrOrZr[rKrKrLr_ r`zPubSub.__aexit__cs|jjdIdHS)NrWrr@execute_pubsub_methodrQrKrKrLrW sz PubSub.aclosecCs |jjjjSrO)rr@ active_pubsub subscribedrQrKrKrLrs zPubSub.subscribedrcs|jjjdg|RIdHS)NrrrJrrKrKrLrs zPubSub.execute_commandrc$|jjjdg|Ri|IdHS)aE Subscribe to channel patterns. Patterns supplied as keyword arguments expect a pattern name as the key and a callable as the value. A pattern's callable will be invoked automatically when a message is received on that pattern rather than producing a message via ``listen()``. psubscribeNrrrKrKrLr zPubSub.psubscribec|jjjdg|RIdHS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscribeNrrrKrKrLr% zPubSub.punsubscribecr)aR Subscribe to channels. Channels supplied as keyword arguments expect a channel name as the key and a callable as the value. A channel's callable will be invoked automatically when a message is received on that channel rather than producing a message via ``listen()`` or ``get_message()``. subscribeNrrrKrKrLr.rzPubSub.subscribecr)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscribeNrrrKrKrLr:rzPubSub.unsubscribeFignore_subscribe_messagestimeoutcs|jjjd||dIdHS)a Get the next message if one is available, otherwise None. If timeout is specified, the system will wait for `timeout` seconds before returning. Timeout should be specified as a floating point number or None to wait indefinitely. get_message)rrNr)rJrrrKrKrLrCs   zPubSub.get_messageg?)exception_handler poll_timeoutrcs|jjj|||dIdHS)aProcess pub/sub messages using registered callbacks. This is the equivalent of :py:meth:`redis.PubSub.run_in_thread` in redis-py, but it is a coroutine. To launch it as a separate task, use ``asyncio.create_task``: >>> task = asyncio.create_task(pubsub.run()) To shut it down, use asyncio cancellation: >>> task.cancel() >>> await task ) sleep_timerrN)rr@execute_pubsub_run)rJrrrKrKrLrunSs z PubSub.run)rNrr)Fr)rrrrr#rMrRr_rWpropertyrrr rrrrrrrrrrrrrKrKrKrLrs4      r);rBloggingtypingrrrrrrredis.asyncio.clientr&redis.asyncio.multidb.command_executorr redis.asyncio.multidb.configr r r r redis.asyncio.multidb.databaserrr&redis.asyncio.multidb.failure_detectorr!redis.asyncio.multidb.healthcheckrrredis.asyncio.retryrredis.backgroundr redis.backoffrredis.commandsrrredis.multidb.circuitrrriredis.multidb.exceptionrrrredis.observability.attributesr redis.typingrr r! redis.utilsr" getLoggerrrr#rrrrKrKrKrLs8            D