File: //usr/local/lib/python3.9/site-packages/kombu/__pycache__/mixins.cpython-39.pyc
a
X>h�% � @ s� d Z ddlmZ ddlZddlmZ ddlmZ ddlm Z ddl
m Z dd l m
Z
dd
lmZ dd lmZmZ dd lmZ dd
lmZ ddlmZ ddlmZ dZee �Z e j e j e j e j!f\Z Z Z"Z!dZ#dZ$G dd� d�Z%G dd� de%�Z&dS )zMixins.� )� annotationsN)�contextmanager)�partial)�count)�sleep� ��
ignore_errors)�
get_logger)�Consumer�Producer��nested)� safe_repr�� TokenBucket)�cached_property)�
ConsumerMixin�ConsumerProducerMixinz?Connection to broker lost, trying to re-establish connection...z8Broker connection error, trying again in %s seconds: %r.c @ s� e Zd ZdZdZdZdd� Zdd� Zdd � Zd
d � Z d d
� Z
dd� Z dd� Z e
dd� �Zd+dd�Ze
dd� �Zd,dd�Zdd � Zd d � Ze
d d � �Ze
d!d"� �Zd#d$� Zed%d&� �Zed'd(� �Zed)d*� �ZdS )-r a' 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()
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 canceled.
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.
NFc C s t d��d S )NzSubclass responsibility)�NotImplementedError)�selfr �channel� r �6/usr/local/lib/python3.9/site-packages/kombu/mixins.py�
get_consumers� s zConsumerMixin.get_consumersc C s d S �Nr �r r r r �on_connection_revived� s z#ConsumerMixin.on_connection_revivedc K s d S r r )r �
connectionr � consumers�kwargsr r r �on_consume_ready� s z ConsumerMixin.on_consume_readyc C s d S r r �r r r r r r �on_consume_end� s z ConsumerMixin.on_consume_endc C s d S r r r r r r � on_iteration� s zConsumerMixin.on_iterationc C s& t d||j|jt|j�� |�� d S )Nz;Can't decode message body: %r (type:%r encoding:%r raw:%r'))�error� content_type�content_encodingr �bodyZack)r �message�excr r r �on_decode_error� s
�z ConsumerMixin.on_decode_errorc C s t t||dd� d S )Nr ��exc_info)�warn� W_CONN_ERROR)r r* �intervalr r r �on_connection_error� s z!ConsumerMixin.on_connection_errorc c s
d V d S r r r" r r r �
extra_context� s zConsumerMixin.extra_contextr c K sz | j }| jj| jj }| jsvz8|�|�rD| jf dd i|��D ]}q<nt|�|�� W q |yr t t
dd� Y q0 qd S )N�limitr r, ) �
restart_limitr �connection_errors�channel_errors� should_stopZ can_consume�consumer Z
expected_timer. � W_CONN_LOST)r Z_tokensr r4 �errors�_r r r �run� s �
zConsumerMixin.runc k s� | � � �d\}}}| �||��2 | j|||fi |�� |||fV W d � n1 sT0 Y W d � n1 sr0 Y d S r )r r2 r! )r r r r r r r r �consumer_context� s z ConsumerMixin.consumer_contextc
k s� d}| j f i |����\}}}|r(t|�p,t� D ]|} | jr< q�| �� z|j|d� W nJ tjy� |�� ||7 }|r�||kr�� Y q. t y� | js�� Y q.0 d V d}q.W d � n1 s�0 Y t
d� d S )Nr )�timeoutzconsume exiting) r= �ranger r7 r$ Z drain_events�socketr> Zheartbeat_check�OSError�debug)
r r3 r> Zsafety_intervalr �elapsed�connr r �ir r r r8 � s& $zConsumerMixin.consumec C s
t | |�S )z/Use :func:`kombu.common.ignore_errors` instead.r )r Zfunr r r �maybe_conn_error� s z ConsumerMixin.maybe_conn_errorc C s
| j �� S r )r �cloner r r r �create_connection� s z ConsumerMixin.create_connectionc c sB | � � �&}|�| j| j� |V W d � n1 s40 Y d S r )rH �ensure_connectionr1 �connect_max_retries�r rD r r r �establish_connection� s
�z"ConsumerMixin.establish_connectionc c s� | � � ��}| �� td|�� � |j}tt|| jd�}| j| � ||�� � }|||fV W d � n1 sj0 Y t
d� | � ||� W d � n1 s�0 Y t
d� d S )NzConnected to %s)r+ zConsumers canceledzConnection closed) rL r �info�as_uriZdefault_channelr r r+ �
_consume_fromr rB r# )r rD r �cls�cr r r r � s
�**zConsumerMixin.Consumerc G s t |� S r r
)r r r r r rO � s zConsumerMixin._consume_fromc C s t d�S )Nr r r r r r r4 � s zConsumerMixin.restart_limitc C s | j jS r )r r5 r r r r r5 � s z ConsumerMixin.connection_errorsc C s | j jS r )r r6 r r r r r6 � s z ConsumerMixin.channel_errors)r )NNr )�__name__�
__module__� __qualname__�__doc__rJ r7 r r r! r# r$ r+ r1 r r2 r< r= r8 rF rH rL r rO r r4 r5 r6 r r r r r & s: ]
r c @ s4 e Zd ZdZdZdd� Zedd� �Zedd� �ZdS ) r a Consumer and Producer mixin.
Version of ConsumerMixin having separate connection for also
publishing messages.
Example:
-------
.. code-block:: python
class Worker(ConsumerProducerMixin):
def __init__(self, connection):
self.connection = connection
def get_consumers(self, Consumer, channel):
return [Consumer(queues=Queue('foo'),
on_message=self.handle_message,
accept='application/json',
prefetch_count=10)]
def handle_message(self, message):
self.producer.publish(
{'message': 'hello to you'},
exchange='',
routing_key=message.properties['reply_to'],
correlation_id=message.properties['correlation_id'],
retry=True,
)
Nc C s | j d ur| j �� d | _ d S r )�_producer_connection�closer" r r r r# s
z$ConsumerProducerMixin.on_consume_endc C s
t | j�S r )r �producer_connectionr r r r �producer$ s z ConsumerProducerMixin.producerc C s0 | j d u r*| j�� }|�| j| j� || _ | j S r )rV r rG rI r1 rJ rK r r r rX ( s
�z)ConsumerProducerMixin.producer_connection) rR rS rT rU rV r# �propertyrY rX r r r r r � s
r )'rU �
__future__r r@ �
contextlibr � functoolsr � itertoolsr �timer �commonr �logr
Z messagingr r Z utils.compatr Zutils.encodingr Z utils.limitsr Z
utils.objectsr �__all__rR �loggerrB rM �warningr% r. r9 r/ r r r r r r �<module> s2 � Y