P'^c@@sdZddlmZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl m Z ddlmZmZmZdd lmZdd lmZdd lmZdd lmZdd lmZmZddlmZddlm Z m!Z!m"Z"ddddgZ#edddZ$e ddZ%dZ&dZ'dZ(de*ejdZ+dZ,de-fdYZ.defdYZ/de-fd YZ0dS(!s celery.events ~~~~~~~~~~~~~ Events is a stream of messages sent for certain actions occurring in the worker (and clients if :setting:`CELERY_SEND_TASK_SENT_EVENT` is enabled), used for monitoring purposes. i(tabsolute_importN(tdeque(tcontextmanager(tcopy(t itemgetter(tExchangetQueuetProducer(t maybe_channel(t ConsumerMixin(tcached_property(tapp_or_default(t anon_nodenametuuid(t dictfilter(tadjust_timestampt utcoffsett maybe_s_to_mstEventstEventtEventDispatchert EventReceivertceleryevttypettopicRt timestamps anyjson is currently using the yajl library. This json implementation is broken, it severely truncates floats so timestamps will not work. Please uninstall yajl or force anyjson to use a different library. icC@s.tt}|jjdkr*d|_n|S(Ntredistfanout(Rtevent_exchanget transportt driver_typeR(tconntex((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt get_exchange2s  cK@sQ|r|||n|}d|krC|jd|d|n ||d<|S(sCreate an event. An event is a dictionary, the only required field is ``type``. A ``timestamp`` field will be set to the current time if not provided. RR(tupdate(Rt_fieldst__dict__t__now__tfieldstevent((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyR:s   cC@s|jdddS(sGet the group part of an event type name. E.g.:: >>> group_from('task-sent') 'task' >>> group_from('custom-my-event') 'custom' t-ii(tsplit(R((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt group_fromIs c B@seZdZedgZdZdZdZdde de ddddZ dZ dZ dZ dZdZedeeedZed Zd Zd Zd Zd ZdZeeeZRS(sDispatches event messages. :param connection: Connection to the broker. :keyword hostname: Hostname to identify ourselves as, by default uses the hostname returned by :func:`~celery.utils.anon_nodename`. :keyword groups: 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"``. :keyword enabled: Set to :const:`False` to not actually publish any events, making :meth:`send` a noop operation. :keyword channel: Can be used instead of `connection` to specify an exact channel to use when sending events. :keyword buffer_while_offline: If enabled events will be buffered while the connection is down. :meth:`flush` must be called as soon as the connection is re-established. You need to :meth:`close` this after use. tsqlc C@st|p|j|_||_||_|p6t|_||_tj|_ d|_ t |_ |p{|jjj|_t|_t|_t|pg|_tj tj g|_|jj|_| r|r|jj|_n||_|jp|jj} t| |_| jj|j krDt!|_n|jrZ|j"ni|jd6|_#t$j%|_&|j'dS(Nthostname((R tappt connectiontchannelR R,tbuffer_while_offlinet threadingtLocktmutextNonetproducerRt_outbound_buffertconftCELERY_EVENT_SERIALIZERt serializertsett on_enabledt on_disabledtgroupsttimettimezonetaltzonettzoffsettclocktclienttenabledR!texchangeRRtDISABLED_TRANSPORTStFalsetenabletheaderstostgetpidtpidt warn_if_yajl( tselfR.R,RDR/R0R-R9R=tconninfo((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt__init__}s4            cC@s8ddl}|jjdkr4tjttndS(Nityajl(tanyjsontimplementationtnametwarningstwarnt UserWarningtW_YAJL(RNRR((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRMs cC@s|S(N((RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt __enter__scG@s|jdS(N(tclose(RNtexc_info((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt__exit__scC@sUt|jp|jd|jd|j|_t|_x|jD] }|q@WdS(NRER9( RR/R.RER9R5tTrueRDR;(RNtcallback((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRHs   cC@s>|jr:t|_|jx|jD] }|q&WndS(N(RDRGRZR<(RNR^((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytdisables    c C@s|j|rdn |jj} ||d|jd|d|jd| |} |j} |j| d|jddd| j d |d |d | gd |j d |j WdQXdS(s8Publish event using a custom :class:`~kombu.Producer` instance. :param type: Event type name, with group separated by dash (`-`). :param fields: Dictionary of event fields, must be json serializable. :param producer: :class:`~kombu.Producer` instance to use, only the ``publish`` method will be called. :keyword retry: Retry in the event of connection failure. :keyword retry_policy: Dict of custom retry policy, see :meth:`~kombu.Connection.ensure`. :keyword blind: Don't set logical clock value (also do not forward the internal logical clock). :keyword Event: Event type used to create event, defaults to :func:`Event`. :keyword utcoffset: Function returning the current utcoffset in hours. R,RRLRBt routing_keyR(t.REtretryt retry_policytdeclareR9RIN( R3R4RBtforwardR,RLREtpublishtreplaceRTR9RI( RNRR&R5RbRctblindRRRBR'RE((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRfs     cK@s|jr|j}|r.t||kr.dSy|j|||j|Wqtk r}|jsln|jj|||fqXndS(sySend event. :param type: Event type name, with group separated by dash (`-`). :keyword retry: Retry in the event of connection failure. :keyword retry_policy: Dict of custom retry policy, see :meth:`~kombu.Connection.ensure`. :keyword blind: Don't set logical clock value (also do not forward the internal logical clock). :keyword Event: Event type used to create event, defaults to :func:`Event`. :keyword utcoffset: Function returning the current utcoffset in hours. :keyword \*\*fields: Event fields, must be json serializable. N( RDR=R*RfR5t ExceptionR0R6tappend(RNRRhR&R=texc((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytsends   cC@sUxN|jrPy|jj\}}}Wntk r<dSX|j||qWdS(sFlushes the outbound buffer.N(R6tpopleftt IndexErrorRl(RNRR&t_((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytflushs   cC@s|jj|jdS(s/Copies the outbound buffer of another instance.N(R6textend(RNtother((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt extend_bufferscC@s)|jjo|jjd|_dS(sClose the event dispatcher.N(R3tlockedtreleaseR4R5(RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRZscC@s|jS(N(R5(RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt_get_publisherscC@s ||_dS(N(R5(RNR5((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt_set_publisher sN(t__name__t __module__t__doc__R:RFR4R-R;R<R]RPRMRYR\RHR_RGRRRfRlRpRsRZRvRwtpropertyt publisher(((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRXs.         #     cB@seZdZdZdddddddZdZdZdZe dZ dde dZ dde d Z dd Z e ejeeed Zd Zed ZRS(sCapture events. :param connection: Connection to the broker. :keyword handlers: Event handlers. :attr:`handlers` is a dict of event types and their handlers, the special handler `"*"` captures all events that doesn't have a handler. t#Rc C@s9t|p|j|_t||_|dkr9in||_||_|pWt|_||_ t |j p~|jj |_ t dj|j |jgd|j d|jdtdtd|j|_|jj|_|jj|_|jj|_|dkr,t|jjjdg}n||_dS(NRaRER`t auto_deletetdurabletqueue_argumentstjson(R R-RR/R4thandlersR`R tnode_idt queue_prefixR!R.RERtjoinR]RGt_get_queue_argumentstqueueRBtadjustt adjust_clockRet forward_clockR:R7R8taccept(RNR/RR`RR-RR((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRPs&  !   cC@s6|jj}tit|jd6t|jd6S(Ns x-message-ttls x-expires(R-R7RRtCELERY_EVENT_QUEUE_TTLtCELERY_EVENT_QUEUE_EXPIRES(RNR7((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyR2s cC@s8|jj|p!|jjd}|o3||dS(sPProcess the received event by dispatching it to the appropriate handler.t*N(Rtget(RNRR'thandler((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytprocess9s$c C@s1|d|jgd|jgdtd|jgS(Ntqueuest callbackstno_ackR(Rt_receiveR]R(RNtConsumerR/((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt get_consumers?scK@s|r|jd|ndS(NR/(twakeup_workers(RNR.R/t consumerstwakeuptkwargs((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyton_consume_readyDscC@s|jd|d|d|S(NtlimitttimeoutR(tconsume(RNRRR((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt itercaptureIscC@s"t|jd|d|d|S(sOpen up a consumer capturing events. This has to run in the main process, and it will never stop unless :attr:`EventDispatcher.should_stop` is set to True, or forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`. RRR(tlistR(RNRRR((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytcaptureLscC@s&|jjjdd|jd|dS(Nt heartbeatR.R/(R-tcontrolt broadcastR.(RNR/((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRVs c C@s|d}|dkrD|jjp%d|}|d<|j|n?y|d} Wn!tk ru|j|dt _TZGETTERRRRRR{R.(((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRs$           cB@sYeZddZedZedZedZede e dZ RS(cC@s ||_dS(N(R-(RNR-((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyRPscC@s|jjtddS(Ntreversesevents.Receiver(R-tsubclass_with_selfR(RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytReceiverscC@s|jjtddS(NRsevents.Dispatcher(R-RR(RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt DispatcherscC@s|jjdddS(Nscelery.events.state:StateRs events.State(R-R(RN((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytStatesc c@sW|jjjjdt6}|j|j|||j| }|VWdQXWdQXdS(Ntblock(R-tamqpt producer_pooltacquireR]RR.R/(RNR,RDR0tprodtd((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pytdefault_dispatchersN( RxRyR4RPR RRRRR]RGR(((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyR~s (1Rzt __future__RRJR>R1RUt collectionsRt contextlibRRtoperatorRtkombuRRRtkombu.connectionRt kombu.mixinsR t kombu.utilsR t celery.appR t celery.utilsR R tcelery.utils.functionalRtcelery.utils.timeutilsRRRt__all__RRRXRR!R4tdictRR*tobjectRRR(((s8/tmp/pip-unpacked-wheel-gV1wwp/celery/events/__init__.pyt s8      p