
    ^f	&                    <   d Z ddlmZ ddlZddlmZ ddlmZ ddlm	Z	 ddl
mZ dd	lmZ dd
lmZ ddlmZmZ ddlmZ ddlmZ ddlmZ ddlmZ dZ ee          Zej        ej        ej         ej!        f\  ZZZ"Z!dZ#dZ$ G d d          Z% G d de%          Z&dS )zMixins.    )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                      e Zd ZdZdZdZd Zd Zd Zd Z	d Z
d	 Zd
 Zed             ZddZed             ZddZd Zd Zed             Zed             Zd Zed             Zed             Zed             ZdS )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                     t          d          )NzSubclass responsibility)NotImplementedError)selfr   channels      >/var/www/html/env/lib/python3.11/site-packages/kombu/mixins.pyget_consumerszConsumerMixin.get_consumers   s    !";<<<    c                    d S N r   s    r   on_connection_revivedz#ConsumerMixin.on_connection_revived       r   c                    d S r   r    )r   
connectionr   	consumerskwargss        r   on_consume_readyzConsumerMixin.on_consume_ready   r#   r   c                    d S r   r    r   r%   r   s      r   on_consume_endzConsumerMixin.on_consume_end   r#   r   c                    d S r   r    r!   s    r   on_iterationzConsumerMixin.on_iteration   r#   r   c           	         t          d||j        |j        t          |j                             |                                 d S )Nz;Can't decode message body: %r (type:%r encoding:%r raw:%r'))errorcontent_typecontent_encodingr   bodyack)r   messageexcs      r   on_decode_errorzConsumerMixin.on_decode_error   sD    K7')A%%	' 	' 	' 	r   c                6    t          t          ||d           d S )Nr   exc_info)warnW_CONN_ERROR)r   r5   intervals      r   on_connection_errorz!ConsumerMixin.on_connection_error   s    \8S1555555r   c              #     K   d V  d S r   r    r*   s      r   extra_contextzConsumerMixin.extra_context   s      r   r   c                @   | j         }| j        j        | j        j        z   }| j        sv	 |                    |          r | j        ddd i|D ]}n"t          |                    |                     n!# |$ r t          t          d           Y nw xY w| j        td S d S )Nlimitr   r8   r    )restart_limitr%   connection_errorschannel_errorsshould_stopcan_consumeconsumer   expected_timer:   W_CONN_LOST)r   _tokensr'   rB   errors_s         r   runzConsumerMixin.run   s    */3/01" 	.. ,,W55 @)T\?????   -55g>>??? . . .[1------. " 	. 	. 	. 	. 	.s   A
A4 4BBc              +     K   |                                  5 \  }}}|                     ||          5   | j        |||fi | |||fV  d d d            n# 1 swxY w Y   d d d            d S # 1 swxY w Y   d S r   )r   r?   r(   )r   r'   r%   r   r&   s        r   consumer_contextzConsumerMixin.consumer_context   s     ]]__ 	5 @Wi##J88 5 5%%j'9OOOOO '944445 5 5 5 5 5 5 5 5 5 5 5 5 5 5	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5 	5s4   A.A
A.A	A.A	A..A25A2c              +    K   d} | j         di |5 \  }}}|rt          |          pt                      D ]}	| j        r n||                                  	 |                    |           d V  d}=# t          j        $ r% |r |                                 ||z  }||k    r Y pt          $ r | j        s Y w xY wd d d            n# 1 swxY w Y   t          d           d S )Nr   )timeoutzconsume exitingr    )rO   ranger   rE   r-   drain_eventssocketrQ   heartbeat_checkOSErrordebug)
r   rA   rQ   safety_intervalr'   elapsedconnr   r&   is
             r   rG   zConsumerMixin.consume   s     "T",,V,, 	 0Jw	+uU||6uww    # E!!### %%o%>>> EEEGG ~ " " " ",,...?2"g--!   +  	  	  	  	  	  	  	  	  	  	  	  	  	  	  	 & 	     sB   AC	A3,C	31B:$C	&B:7C	9B::C		CCc                "    t          | |          S )z/Use :func:`kombu.common.ignore_errors` instead.r	   )r   funs     r   maybe_conn_errorzConsumerMixin.maybe_conn_error   s    T3'''r   c                4    | j                                         S r   )r%   cloner!   s    r   create_connectionzConsumerMixin.create_connection   s    $$&&&r   c              #     K   |                                  5 }|                    | j        | j                   |V  d d d            d S # 1 swxY w Y   d S r   )ra   ensure_connectionr=   connect_max_retriesr   rZ   s     r   establish_connectionz"ConsumerMixin.establish_connection   s      ##%% 	""4#;#'#;= = =JJJ	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	 	s   %A		AAc              #    K   |                                  5 }|                                  t          d|                                           |j        }t          t          || j                  } | j        | 	                    ||           5 }|||fV  d d d            n# 1 swxY w Y   t          d           |                     ||           d d d            n# 1 swxY w Y   t          d           d S )NzConnected to %s)r6   zConsumers canceledzConnection closed)rf   r"   infoas_uridefault_channelr   r   r6   _consume_fromr   rW   r+   )r   rZ   r   clscs        r   r   zConsumerMixin.Consumer   s     &&(( 		/D&&((("DKKMM222*G(G*.*>@ @ @C##T%7%7W%E%EF '!GQ&&&&' ' ' ' ' ' ' ' ' ' ' ' ' ' '&'''g...		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 		/ 	!"""""s6   A8CB#C#B'	'C*B'	+(CC#&C#c                    t          | S r   r   )r   r&   s     r   rk   zConsumerMixin._consume_from   s    y!!r   c                     t          d          S )Nr   r   r!   s    r   rB   zConsumerMixin.restart_limit   s    1~~r   c                    | j         j        S r   )r%   rC   r!   s    r   rC   zConsumerMixin.connection_errors   s    00r   c                    | j         j        S r   )r%   rD   r!   s    r   rD   zConsumerMixin.channel_errors   s    --r   )r   )NNr   )__name__
__module____qualname____doc__rd   rE   r   r"   r(   r+   r-   r6   r=   r   r?   rM   rO   rG   r^   ra   rf   r   rk   r   rB   rC   rD   r    r   r   r   r   &   s       Y Yz 
 K= = =          6 6 6   ^. . . . 5 5 ^5! ! ! !.( ( (' ' '   ^ # # ^#" " "   _ 1 1 _1 . . _. . .r   r   c                  H    e Zd ZdZdZd Zed             Zed             ZdS )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                X    | j         "| j                                          d | _         d S d S r   )_producer_connectioncloser*   s      r   r+   z$ConsumerProducerMixin.on_consume_end   s6    $0%++---(,D%%% 10r   c                *    t          | j                  S r   )r   producer_connectionr!   s    r   producerzConsumerProducerMixin.producer%  s    0111r   c                    | j         @| j                                        }|                    | j        | j                   || _         | j         S r   )rx   r%   r`   rc   r=   rd   re   s     r   r{   z)ConsumerProducerMixin.producer_connection)  sR    $,?((**D""4#;#'#;= = =(,D%((r   )	rr   rs   rt   ru   rx   r+   propertyr|   r{   r    r   r   r   r      sk         <  - - -
 2 2 X2 ) ) X) ) )r   r   )'ru   
__future__r   rT   
contextlibr   	functoolsr   	itertoolsr   timer   commonr
   logr   	messagingr   r   utils.compatr   utils.encodingr   utils.limitsr   utils.objectsr   __all__rr   loggerrW   rh   warningr/   r:   rI   r;   r   r   r    r   r   <module>r      s    " " " " " "  % % % % % %                   ! ! ! ! ! !       ) ) ) ) ) ) ) )             % % % % % % % % % % % % * * * * * *
4	H		
L
K
N
L	 tT5
V. V. V. V. V. V. V. V.r1) 1) 1) 1) 1)M 1) 1) 1) 1) 1)r   