(i6ddlZddlZddlmZmZmZmZmZmZm Z m Z ddl m Z ddl mZmZmZmZmZddlmZddlmZmZddlmZmZerddlmZdd lmZmZGd d Z y) N) TYPE_CHECKINGAnyListSequenceSetUnioncastoverload)HexStr) SubscriptionHandlerTaskExceptionSubscriptionProcessingFinishedTaskNotRunning Web3TypeErrorWeb3ValueError)SubscriptionContainer) FormattedEthSubscriptionResponse RPCResponse)EthSubscriptionEthSubscriptionContext) AsyncWeb3)PersistentConnectionProviderRequestProcessorcDeZdZUdZej dZejed<ddZ de e ddfdZ de e ddfd Z de e ddfd Zdd Zdd Zedee e fd Zdede e fdZdede e fdZede e defdZedee e deefdZdee e ee e fdeeeeffdZede e defdZededefdZedeee e efdefdZdee e eeee e effdefdZdefdZd deddfdZy)!SubscriptionManagerz The ``SubscriptionManager`` is responsible for subscribing, unsubscribing, and managing all active subscriptions for an ``AsyncWeb3`` instance. It is also used for processing all subscriptions that have handler functions. z.web3.providers.persistent.subscription_managerloggerreturnNc||_td|j|_t |_d|_d|_t|_ |j |jj_d|_ y)NrFr) _w3r provider _providerr_subscription_container parallelize task_timeoutset_tasks_request_processortotal_handler_calls)selfw3s _/var/www/br/venv/lib/python3.12/site-packages/web3/providers/persistent/subscription_manager.py__init__zSubscriptionManager.__init__8si$!14  ( ( ))A)* subscriptionc:|jj|yN)r"add_subscriptionr)r.s r+_add_subscriptionz%SubscriptionManager._add_subscriptionKs $$55lCr-c:|jj|yr0)r"remove_subscriptionr2s r+_remove_subscriptionz(SubscriptionManager._remove_subscriptionNs $$88Fr-cV|j|jk(rVd}|j|j8|jd||_|dz }|j|j7yy|j|jj vrt d|jy)N#rzUSubscription label already exists. Subscriptions must have unique labels. label: )label_default_label get_by_label_labelr"subscriptions_by_labelr)r)r.is r+_validate_and_normalize_labelz1SubscriptionManager._validate_and_normalize_labelQs   !+>*?AGr-c |jr>|js- |j|jj |yyy#t $rh}|j jd|jjjjt|t|Yd}~yd}~wwxYw)z Callback when a handler task completes. Similar to _message_listener_callback. Puts handler exceptions into the queue to be raised in the main loop, else removes the task from the set of active tasks. z.Subscription handler task raised an exception.)messageN)done cancelledresultr&discard Exceptionr exceptionr!r'_handler_subscription_queue put_nowaitr str)r)taskes r+_handler_task_callbackz*SubscriptionManager._handler_task_callbackcs 99;t~~/   ##D) 0;  %%&VW11MMXX4T3q6J s+A CAB<<CcK|jsy|jjdt|j|jD]#}|j r|j %|jj yw)z(Cancel and clean up all remaining tasks.Nz!Cleaning up %d remaining tasks...)r&rdebuglenrCcancelclear)r)rLs r+_cleanup_remaining_tasksz,SubscriptionManager._cleanup_remaining_taskssse{{  =s4;;?OPKK D99;   s AB -B c.|jjSr0)r" subscriptions)r)s r+rVz!SubscriptionManager.subscriptionss++999r-sub_idc8|jj|Sr0)r" get_by_id)r)rWs r+rYzSubscriptionManager.get_by_ids++55f==r-r:c8|jj|Sr0)r"r<)r)r:s r+r<z SubscriptionManager.get_by_labels++88??r-rVc Kywr0r)rVs r+ subscribezSubscriptionManager.subscribe  c Kywr0r\r]s r+r^zSubscriptionManager.subscribes  r`cKt|tr||_|j||jj j |jd{}||_|j||jjd|j||St|trLt|dk(r tdg}|D]*}|j!|j#|d{,|St%d77w)z Used to subscribe to a single or multiple subscriptions. :param subscriptions: A single subscription or a sequence of subscriptions. :type subscriptions: Union[EthSubscription, Sequence[EthSubscription]] :return: NzASuccessfully subscribed to subscription: label: %s id: %srNo subscriptions provided.z7Expected a Subscription or a sequence of Subscriptions.) isinstancermanagerr@reth _subscribesubscription_params_idr3rinfor:rrQrappendr^r)r)rVrWsub_idssubs r+r^zSubscriptionManager.subscribes m_ 5$(M !  . .} =2488<<22M4U4UVVF &M   " "= 1 KK  U##  M  x 0=!Q&$%ABB$&G$ :T^^C%889 :NUVV#W 9s%AD DBD 0D 1D  D c Kywr0r\r]s r+ unsubscribezSubscriptionManager.unsubscriber_r`c Kywr0r\r]s r+rozSubscriptionManager.unsubscriber_r`c Kywr0r\r]s r+rozSubscriptionManager.unsubscribes r`c2Kt|tst|trDt|tr#|}|j|}|t d|||j vr%t d|j d|j|jjj|jd{r1|j||jjd|j |jt|j j"dk(rA|j$j&j(}|j+t-d{yt|t.r{t|dk(r t dg}t1|}|D]E}t|tr t3|}|j5|j7|d{Gt9|S|jj;d |y 7T777w) a Used to unsubscribe from one or multiple subscriptions. :param subscriptions: The subscription(s) to unsubscribe from. :type subscriptions: Union[EthSubscription, Sequence[EthSubscription], HexStr, Sequence[HexStr]] :return: ``True`` if unsubscribing to all was successful, ``False`` otherwise with a warning. :rtype: bool NzTSubscription not found or is not being managed by the subscription manager. id: zWSubscription not found or is not being managed by the subscription manager. label: z id: zESuccessfully unsubscribed from subscription: label: %s id: %srTrcz;Failed to unsubscribe from subscription subscription=%sF)rdrrKrYrrVr:rirrf _unsubscribeidr6rrjrQr"handler_subscriptionsr!r'rIputr rlistr rkroallwarning)r)rVsubscription_idqueue unsubscribedsubsrms r+rozSubscriptionManager.unsubscribes$ m_ 5MSV9W--"/ $ ?  ((::I9JL D$6$66$+112*]=N=N&&  ( s&B.B)B' B)A(B.'B))B. run_forevercK|jjs|s|jjdy|jj j }|s|jjrk tt|jd{}tt|jjj|d{}|jd}|jj|}|rt|j||dfi|j }|j"dus|j"g|j"r[t%j&|j)|}|j*j-||j/|j0n|j)|d{|rS|jjrk|jj jC|j*r|j;d{yy77_7p#t2$r"|s|jj5dYzYt6$r6|jj9d|j;d{7t<$rK} |jj9d| j>|jjAYd} ~ d} ~ wwxYw7˭w) a{ Used to handle all subscriptions that have handlers. The method will run until all subscriptions that have handler functions are unsubscribed from or, if ``run_forever`` is set to ``True``, it will run indefinitely. :param run_forever: If ``True``, the method will run indefinitely. :type run_forever: bool :return: None zANo handler subscriptions found. Subscription handler did not run.Nr.rETzVAll handler subscriptions have been unsubscribed from. Stopping subscription handling.zUAn exception occurred in a subscription handler task. Stopping subscription handling.z"Stopping subscription handling: %s)"r"rurryr!r'rIr rgetrrre_process_responseget_handler_subscription_by_idr_handler_contextr#asyncio create_task_handlerr&addadd_done_callbackrNr rjr errorrTrrB _handle_listener_task_exceptions!_reset_handler_subscription_queue) r)rr{responseformatted_sub_responserWrm sub_contextrLrMs r+handle_subscriptionsz(SubscriptionManager.handle_subscriptions"sm++AA+ KK  S  11MMT99OO0  599;->?)-4((**<>?3+?>@#@/#*>@ _S-A f   %oc&:; f  W"?3#7/RUBV9W#WXW vtF|# $W@ s/C    v $   oc&:F&B CD    E C  U?3/78 9 ; E ENt2H2dH2tH2r-r)!rrtypingrrrrrrr r eth_typingr web3.exceptionsr r rrr0web3.providers.persistent.subscription_containerr web3.typesrrweb3.utils.subscriptionsrrweb3rweb3.providers.persistentrrrr\r-r+rsQ     }2}2r-