O'^c@@s=dZddlmZddlZddlmZddlmZddlm Z ddl m Z dd l m Z dd lmZdd lmZdd lmZdd lmZmZddlmZddlmZdgZeeZejejej ej!f\ZZZ Z!dZ"dZ#de$fdYZ%dS(s3 kombu.mixins ============ Useful mixin classes. i(tabsolute_importN(tcontextmanager(tpartial(tcount(tsleepi(t ignore_errors(trange(tConsumer(t get_logger(tcached_propertytnested(t safe_repr(t TokenBuckett ConsumerMixins?Connection to broker lost, trying to re-establish connection...s8Broker connection error, trying again in %s seconds: %r.cB@seZdZdZeZdZdZdZ dZ dZ dZ dZ edZd d Zed Zddd d Zd ZdZedZedZdZedZedZedZRS(s Convenience mixin for implementing consumer programs. It can be used outside of threads, with threads, or greenthreads (eventlet/gevent) too. The basic class would need a :attr:`connection` attribute which must be a :class:`~kombu.Connection` instance, and define a :meth:`get_consumers` method that returns a list of :class:`kombu.Consumer` instances to use. Supporting multiple consumers is important so that multiple channels can be used for different QoS requirements. **Example**: .. code-block:: python class Worker(ConsumerMixin): task_queue = Queue('tasks', Exchange('tasks'), 'tasks')) def __init__(self, connection): self.connection = None def get_consumers(self, Consumer, channel): return [Consumer(queues=[self.task_queue], callbacks=[self.on_task])] def on_task(self, body, message): print('Got task: {0!r}'.format(body)) message.ack() **Additional handler methods**: * :meth:`extra_context` Optional extra context manager that will be entered after the connection and consumers have been set up. Takes arguments ``(connection, channel)``. * :meth:`on_connection_error` Handler called if the connection is lost/ or is unavailable. Takes arguments ``(exc, interval)``, where interval is the time in seconds when the connection will be retried. The default handler will log the exception. * :meth:`on_connection_revived` Handler called as soon as the connection is re-established after connection failure. Takes no arguments. * :meth:`on_consume_ready` Handler called when the consumer is ready to accept messages. Takes arguments ``(connection, channel, consumers)``. Also keyword arguments to ``consume`` are forwarded to this handler. * :meth:`on_consume_end` Handler called after the consumers are cancelled. Takes arguments ``(connection, channel)``. * :meth:`on_iteration` Handler called for every iteration while draining events. Takes no arguments. * :meth:`on_decode_error` Handler called if a consumer was unable to decode the body of a message. Takes arguments ``(message, exc)`` where message is the original message object. The default handler will log the error and acknowledge the message, so if you override make sure to call super, or perform these steps yourself. cC@stddS(NsSubclass responsibility(tNotImplementedError(tselfRtchannel((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyt get_consumersscC@sdS(N((R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyton_connection_revivedscK@sdS(N((Rt connectionRt consumerstkwargs((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyton_consume_readyscC@sdS(N((RRR((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyton_consume_endscC@sdS(N((R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyt on_iterationscC@s3td||j|jt|j|jdS(Ns;Can't decode message body: %r (type:%r encoding:%r raw:%r')(terrort content_typetcontent_encodingR tbodytack(Rtmessagetexc((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyton_decode_errorscC@stt||dddS(Ntexc_infoi(twarnt W_CONN_ERROR(RRtinterval((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyton_connection_errorscc@s dVdS(N((RRR((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyt extra_contextsicC@s|j}|jj|jj}xw|jsyF|j|r]x0|jddD]}qPWnt|j |Wq"|k rt t ddq"Xq"WdS(NtlimitR!i( t restart_limitRtconnection_errorstchannel_errorst should_stopt can_consumetconsumetNoneRt expected_timeR"t W_CONN_LOST(Rt_tokensR(terrorst_((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pytruns     c k@s_|jM\}}}|j||)|j|||||||fVWdQXWdQXdS(N(RR&R(RRRRR((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pytconsumer_contextsc k@sd}|j|\}}}x|r6t|p<tD]} |jrPPn|jy|jd|Wnbtjk r|j||7}|r||krqq=tj k r|jsqq=XdVd}q=WWdQXt ddS(Nittimeoutsconsume exiting( R5RRR+Rt drain_eventstsocketR6theartbeat_checkRtdebug( RR'R6tsafety_intervalRtelapsedtconnRRti((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyR-s&"       cC@s t||S(s/Use :func:`kombu.common.ignore_errors` instead.(R(Rtfun((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pytmaybe_conn_errorscC@s |jjS(N(Rtclone(R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pytcreate_connectionscc@s4|j"}|j|j|j|VWdQXdS(N(RBtensure_connectionR%tconnect_max_retries(RR=((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pytestablish_connections  c c@s|j}|jtd|j|j}tt|d|j}|j|j ||}|||fVWdQXt d|j ||WdQXt ddS(NsConnected to %sR sConsumers cancelledsConnection closed( RERtinfotas_uritdefault_channelRRR t _consume_fromRR:R(RR=Rtclstc((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyRs     cG@s t|S(N(R (RR((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyRIscC@s tdS(Ni(R (R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyR(scC@s |jjS(N(RR)(R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyR)scC@s |jjS(N(RR*(R((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyR*sN(t__name__t __module__t__doc__R.RDtFalseR+RRRRRR R%RR&R4R5R-R@RBRERRIR R(R)R*(((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyR (s,[           (&RNt __future__RR8t contextlibRt functoolsRt itertoolsRttimeRtcommonRtfiveRt messagingRtlogRtutilsR R tutils.encodingR t utils.limitsR t__all__RLtloggerR:RFR"RR0R#tobjectR (((s./tmp/pip-unpacked-wheel-UAnTfW/kombu/mixins.pyts&   *