h#dZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z ddl mZdd lmZmZmZd ZGd d Zy) zEvent dispatcher sends events.N) defaultdictdeque)Producer)app_or_default) anon_nodename) utcoffset)Event get_exchange group_from)EventDispatcherceZdZdZdhZdZdZdZ ddZdZ dZ dZ dZ d e fd Zd defd Zd ed de fd Zdd ZdZdZdZdZeeeZy)r a0Dispatches event messages. Arguments: connection (kombu.Connection): Connection to the broker. hostname (str): Hostname to identify ourselves as, by default uses the hostname returned by :func:`~celery.utils.anon_nodename`. groups (Sequence[str]): List of groups to send events for. :meth:`send` will ignore send requests to groups not in this list. If this is :const:`None`, all events will be sent. Example groups include ``"task"`` and ``"worker"``. enabled (bool): Set to :const:`False` to not actually publish any events, making :meth:`send` a no-op. channel (kombu.Channel): Can be used instead of `connection` to specify an exact channel to use when sending events. buffer_while_offline (bool): If enabled events will be buffered while the connection is down. :meth:`flush` must be called as soon as the connection is re-established. Note: You need to :meth:`close` this after use. sqlNc ~t|xs |j|_||_||_|xs t |_||_| xs t|_| |_ | |_ tt|_ tj|_d|_t%|_|xs |jj(j*|_t/|_t/|_t/|xsg|_t6j8 t6j: g|_|jj>|_| |_ |s|r|jjB|_||_"|jxs|jjG} tI| |jj(jJ|_&| jNjP|jRvrd|_"|jDr|jUd|j i|_+tYjZ|_.y)N)nameFhostname)/rapp connectionchannelrrbuffer_while_offline frozenset buffer_group buffer_limiton_send_bufferedrlist _group_buffer threadingLockmutexproducerr_outbound_bufferconfevent_serializer serializerset on_enabled on_disabledgroupstimetimezonealtzonetzoffsetclock delivery_modeclientenabledconnection_for_writer event_exchangeexchange transport driver_typeDISABLED_TRANSPORTSenableheadersosgetpidpid)selfrrr0rrrr$r(r.rrrconninfos T/var/www/Befach/backend/env/lib/python3.12/site-packages/celery/events/dispatcher.py__init__zEventDispatcher.__init__:s"#/2$  3MO $8!(7IK( 0(.^^%  %$F (F(F%5&,B' --$,,7 XX^^ *g%0077DO ??Edhh&C&C&E$X*.((--*F*FH    ) )T-E-E E DL << KKM"DMM2 99;c|SNr<s r> __enter__zEventDispatcher.__enter__^s r@c$|jyrB)close)r<exc_infos r>__exit__zEventDispatcher.__exit__as  r@ct|jxs |j|j|jd|_d|_|jD] }| y)NF)r3r$ auto_declareT)rrrr3r$r r0r&r<callbacks r>r7zEventDispatcher.enabledsJ !@*.--,0OO.35  H J(r@c||jr0d|_|j|jD] }| yy)NF)r0rGr'rLs r>disablezEventDispatcher.disablems3 << DL JJL ,, - r@Fc "|rdn|jj}||f|jt|j|d|}|j 5|j ||fd|jddi|cdddS#1swYyxYw)auPublish event using custom :class:`~kombu.Producer`. Arguments: type (str): Event type name, with group separated by dash (`-`). fields: Dictionary of event fields, must be json serializable. producer (kombu.Producer): Producer instance to use: only the ``publish`` method will be called. retry (bool): Retry in the event of connection failure. retry_policy (Mapping): Map of custom retry policy options. See :meth:`~kombu.Connection.ensure`. blind (bool): Don't set logical clock value (also don't forward the internal logical clock). Event (Callable): Event type used to create event. Defaults to :func:`Event`. utcoffset (Callable): Function returning the current utc offset in hours. Nrrr;r- routing_key-.)r-forwardrrr;r_publishreplace) r<typefieldsr blindr kwargsr-events r>publishzEventDispatcher.publishts&4::#5#5#7d;T]]ik((%;39; ZZ O 4==O-1\\#s-COGMO O O Os &BBc &|j} |j|||j|||g|j|j|j  y#t $r5}|js|jj|||fYd}~yd}~wwxYw)N)rRr3retry retry_policydeclarer$r8r.) r3r]rr$r8r. Exceptionrr!append) r<r\r rRr_r`rr3excs r>rVzEventDispatcher._publishs== D   '!)! ?? "00   D,,  ! ! ( (%c)B C C DsAA B+B  Bc |jr|jt|} }|r| |vry| |jvr|jj } ||f|j ||j| d|} |j| } | j| t| |jk\r|jy|jr|jyy|j|||j||||Sy)aSend event. Arguments: type (str): Event type name, with group separated by dash (`-`). retry (bool): Retry in the event of connection failure. retry_policy (Mapping): Map of custom retry policy options. See :meth:`~kombu.Connection.ensure`. blind (bool): Don't set logical clock value (also don't forward the internal logical clock). Event (Callable): Event type used to create event, defaults to :func:`Event`. utcoffset (Callable): unction returning the current utc offset in hours. **fields (Any): Event fields -- must be json serializable. NrQ)rZr r_r`)r0r(r rr-rUrr;rrclenrflushrr]r ) r<rXrZrr_r`r rYr(groupr-r\bufs r>sendzEventDispatcher.sends" << KKD)9EF%v-))) **,dCT]](1 "&((%C;AC((/ 5!s8t000JJL**))++||D&$--u*/u1=$??! r@c|rnt|j} |j5|D]#\}}}|j||j|% ddd|jj |r^|j5|j jD]*\}}|j||jd|zg|dd, dddyy#1swYxYw#|jj wxYw#1swYyxYw)zFlush the outbound buffer.Nz%s.multi)rr!rrVr clearritems) r<errorsr(rir\rR_rheventss r>rgzEventDispatcher.flushs t,,-C .ZZI14-{A eT]]KH25I%%++-  #%)%7%7%=%=%?ME6MM&$--e9KL "F1I&@ # #  II%%++- # #s/ C)CC?AC=CCC:=DcN|jj|jy)z-Copy the outbound buffer of another instance.N)r!extend)r<others r> extend_bufferzEventDispatcher.extend_buffers $$U%;%;rGzEventDispatcher.closes, 4 2 2 4 r@c|jSrBr rDs r>_get_publisherzEventDispatcher._get_publishers }}r@c||_yrBry)r<r s r>_set_publisherzEventDispatcher._set_publishers   r@) NNTNTNNNr NN)TT)__name__ __module__ __qualname____doc__r6rr&r'r?rErIr7rOr r]rrVrjrgrtrGrzr|property publisherrCr@r>r r s8!' CJK?C>B=>FJ"H5O4rs7$  *%0'22 R9R9r@