o Df#@sdZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z ddl mZdd lmZmZmZd ZGd d d ZdS) zEvent dispatcher sends events.N) defaultdictdeque)Producer)app_or_default) anon_nodename) utcoffset)Event get_exchange group_from)EventDispatcherc@seZdZdZdhZdZdZdZ    d"ddZd d Z d d Z d dZ ddZ de fddZddefddZdedde fddZd#ddZddZddZddZd d!ZeeeZdS)$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. sqlNTrc Cs0t|p|j|_||_||_|pt|_||_| pt|_| |_ | |_ t t |_ t|_d|_t|_|p:|jjj|_t|_t|_t|pHg|_tj tj g|_|jj|_| |_ |se|re|jj!|_||_"|jpo|j#} t$| |jjj%d|_&| j'j(|j)vrd|_"|j"r|*d|ji|_+t,-|_.dS)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)selfrrr.rrrr"r&r,rrrconninfor<Q/home/ubuntu/webapp/venv/lib/python3.10/site-packages/celery/events/dispatcher.py__init__:s@       zEventDispatcher.__init__cCs|SNr<r:r<r<r= __enter__^szEventDispatcher.__enter__cGs |dSr?)close)r:exc_infor<r<r=__exit__as zEventDispatcher.__exit__cCs:t|jp|j|j|jdd|_d|_|jD]}|qdS)NF)r1r" auto_declareT)rrrr1r"rr.r$r:callbackr<r<r=r5ds  zEventDispatcher.enablecCs.|jrd|_||jD]}|q dSdS)NF)r.rBr%rFr<r<r=disablems zEventDispatcher.disableFc Ks||rdn|j}||f|jt|j|d|}|j|j||fd|ddi|WdS1s7wYdS)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. Nrrr9r+ routing_key-.)r+forwardrrr9r_publishreplace) r:typefieldsrblindr kwargsr+eventr<r<r=publishts  $zEventDispatcher.publishc Cst|j}z|j|||j|||g|j|j|jd WdSty9}z|js%|j |||fWYd}~dSd}~ww)N)rJr1retry retry_policydeclarer"r6r,) r1rUrr"r6r, Exceptionrrappend) r:rTrrJrVrWrr1excr<r<r=rNs&   zEventDispatcher._publishc Ks|jr\|jt|}} |r| |vrdS| |jvrO|j} ||f|j||j| d|} |j| } | | t | |j krD| dS|j rM| dSdS|j|||j||||dSdS)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. NrI)rRr rVrW)r.r&r rr+rMrr9rrZlenrflushrrUr) r:rPrRrrVrWr rQr&groupr+rTbufr<r<r=sends0        zEventDispatcher.sendc Cs|r8t|j}z*|j|D] \}}}|||j|qWdn1s&wYW|jn|jw|rj|j#|jD]\}}|||jd|g|dd<qCWddS1scwYdSdS)zFlush the outbound buffer.Nz%s.multi)rrrrNrclearritems) r:errorsr&r_rTrJ_r^eventsr<r<r=r]s$ "zEventDispatcher.flushcCs|j|jdS)z-Copy the outbound buffer of another instance.N)rextend)r:otherr<r<r= extend_bufferszEventDispatcher.extend_buffercCs*|jo|jd|_dSd|_dS)zClose the event dispatcher.N)rlockedreleaserr@r<r<r=rBs  zEventDispatcher.closecCs|jSr?rr@r<r<r=_get_publisherszEventDispatcher._get_publishercCs ||_dSr?rk)r:rr<r<r=_set_publishers zEventDispatcher._set_publisher) NNTNTNNNrNrN)TT)__name__ __module__ __qualname____doc__r4rr$r%r>rArDr5rHr rUrrNr`r]rhrBrlrmproperty publisherr<r<r<r=r s: $    %r )rqr7rr' collectionsrrkombur celery.apprcelery.utils.nodenamesrcelery.utils.timerrTr r r __all__r r<r<r<r=s