o wvXjy#@sdZddlmZmZddlZddlZddlZddlmZm Z ddl m Z ddl m Z ddlmZddlmZdd lmZd d lmZmZmZd ZGd ddeZdS)zEvent dispatcher sends events.)absolute_importunicode_literalsN) defaultdictdeque)Producer)app_or_default)items) 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)$ra0Dispatches 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. sqlNTr c 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 connectionchannelr rbuffer_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)selfrrr1rrrr%r)r/rrrconninfor?U/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/events/dispatcher.py__init__<s@       zEventDispatcher.__init__cCs|SNr?r=r?r?r@ __enter__`szEventDispatcher.__enter__cGs |dSrB)close)r=exc_infor?r?r@__exit__cs zEventDispatcher.__exit__cCs:t|jp|j|j|jdd|_d|_|jD]}|qdS)NF)r4r% auto_declareT)rrrr4r%r!r1r'r=callbackr?r?r@r8fs  zEventDispatcher.enablecCs.|jrd|_||jD]}|q dSdS)NF)r1rEr(rIr?r?r@disableos 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. Nrr r<r. routing_key-.)r.forwardrr r<r _publishreplace) r=typefieldsr!blindr kwargsr.eventr?r?r@publishvs  $zEventDispatcher.publishc Cst|j}z|j|||j|||g|j|j|jd WdSty9}z|js%|j |||fWYd}~dSd}~ww)N)rMr4retry retry_policydeclarer%r9r/) r4rXrr%r9r/ Exceptionrr"append) r=rWr!rMrYrZr r4excr?r?r@rQs&   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. NrL)rUr rYrZ)r1r)rrr.rPrr<rr]lenrflushrrXr!) r=rSrUr rYrZr rTr)groupr.rWbufr?r?r@sends0        zEventDispatcher.sendc Cs|r8t|j}z*|j|D] \}}}|||j|qWdn1s&wYW|jn|jw|rj|j#t|jD]\}}|||jd|g|dd<qCWddS1scwYdSdS)zFlush the outbound buffer.Nz%s.multi)rr"r rQr!clearrr) r=errorsr)rbrWrM_raeventsr?r?r@r`s$ "zEventDispatcher.flushcCs|j|jdS)z-Copy the outbound buffer of another instance.N)r"extend)r=otherr?r?r@ extend_bufferszEventDispatcher.extend_buffercCs*|jo|jd|_dSd|_dS)zClose the event dispatcher.N)r lockedreleaser!rCr?r?r@rEs  zEventDispatcher.closecCs|jSrBr!rCr?r?r@_get_publisherszEventDispatcher._get_publishercCs ||_dSrBrm)r=r!r?r?r@_set_publishers zEventDispatcher._set_publisher) NNTNTNNNr NrN)TT)__name__ __module__ __qualname____doc__r7rr'r(rArDrGr8rKr rXr rQrcr`rjrErnroproperty publisherr?r?r?r@rs: $    %r)rs __future__rrr:rr* collectionsrrkombur celery.appr celery.fivercelery.utils.nodenamesr celery.utils.timer rWr r r__all__objectrr?r?r?r@s