File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/redis.cpython-39.pyc
a
X>h� � @ s� d Z ddlmZ ddlZddlZddlZddlmZ ddlmZ ddl m
Z
ddl m Z ddl
mZ dd lmZ dd
lmZ dd lmZ dd lmZmZ dd
lmZ ddlmZ ddlm Z ddl m Z m Z m Z ddl!m"Z" ddl#m$Z$m%Z% ddl&m'Z' ddl(m)Z) ddl*m+Z+ ddl,m-Z- z ddl.Z.ee d��ed�kZ/W n e0�yX dZ.dZ/Y n0 zddl.m1Z1 W n e0�y� dZ1Y n0 ed�Z2e2j3e2j4 Z5Z4d Z6dZ7d Z8g d �Z9ed d �Z:d!d"� Z;d#d$� Z<G d%d&� d&e=�Z>e
d'd(� �Z?d)d*� Z@G d+d,� d,�ZAG d-d.� d.eAe.jB�ZCG d/d0� d0eAe.jDjE�ZFG d1d2� d2e.jDjG�ZHG d3d4� d4e-jI�ZIG d5d6� d6�ZJG d7d8� d8e-jK�ZKG d9d:� d:e-jL�ZLe1�r�G d;d<� d<e1jMe.jN�ZOG d=d>� d>eK�ZPG d?d@� d@eL�ZQdS )Aa� Redis transport module for Kombu.
Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: Yes
* Supports TTL: No
Connection String
=================
Connection string has the following format:
.. code-block::
redis://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]
rediss://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]
To use sentinel for dynamic Redis discovery,
the connection string has following format:
.. code-block::
sentinel://[USER:PASSWORD@]SENTINEL_ADDRESS[:PORT]
Transport Options
=================
* ``sep``
* ``ack_emulation``: (bool) If set to True transport will
simulate Acknowledge of AMQP protocol.
* ``unacked_key``
* ``unacked_index_key``
* ``unacked_mutex_key``
* ``unacked_mutex_expire``
* ``visibility_timeout``
* ``unacked_restore_limit``
* ``fanout_prefix``
* ``fanout_patterns``
* ``global_keyprefix``: (str) The global key prefix to be prepended to all keys
used by Kombu
* ``socket_timeout``
* ``socket_connect_timeout``
* ``socket_keepalive``
* ``socket_keepalive_options``
* ``queue_order_strategy``
* ``max_connections``
* ``health_check_interval``
* ``retry_on_timeout``
* ``priority_steps``
� )� annotationsN)�bisect)�
namedtuple)�contextmanager)�version)�Empty)�time)�Version)�promise)�InconsistencyError�VersionMismatch)�
get_logger)�register_after_fork)� bytes_to_str)�ERR�READ�poll)�accepts_argument)�dumps�loads)�cached_property)�
cycle_by_name)�
_parse_url� )�virtual�redisz5.3.0)�sentinelzkombu.transport.redisi� � )r � � � �error_classes_t)�connection_errors�channel_errorsc
C sb ddl m} t| d�r | j}n| j}ttjjt t
j t t
| j| j| j| jf tjj|| j| jf �S )z$Return tuple of redis error classes.r ��
exceptions� InvalidData)r r% �hasattrr& � DataErrorr! r � Transportr" r �socket�error�IOError�OSError�ConnectionErrorZBusyLoadingError�AuthenticationError� TimeoutErrorr# ZInvalidResponse�
ResponseError)r% r( � r2 �?/usr/local/lib/python3.9/site-packages/kombu/transport/redis.py�get_redis_error_classes} s*
� ��r4 c C s ddl m} | jS )z1Return the redis ConnectionError exception class.r r$ )r r% r. r$ r2 r2 r3 �get_redis_ConnectionError� s r5 c @ s e Zd ZdZdS )� MutexHeldz)Raised when another party holds the lock.N��__name__�
__module__� __qualname__�__doc__r2 r2 r2 r3 r6 � s r6 c c s� | j ||d�}d}zJ|jdd�}|r,dV nt� �W |r�z |�� W q� tjjyZ Y q�0 n*|r�z |�� W n tjjy� Y n0 0 dS )zTAcquire redis lock in non blocking way.
Raise MutexHeld if not successful.
��timeoutF)�blockingN)�lock�acquirer6 �releaser r% ZLockNotOwnedError)�client�nameZexpirer? Z
lock_acquiredr2 r2 r3 �Mutex� s" � rD c C s | � � d S �N)� _after_fork��channelr2 r2 r3 �_after_fork_cleanup_channel� s rI c sl e Zd ZdZg d�Zddd�ddd�ddd�ddd�d �Zd
d � Z� fd d
�Z� fdd�Zddd�Z � Z
S )�GlobalKeyPrefixMixina Mixin to provide common logic for global key prefixing.
Overriding all the methods used by Kombu with the same key prefixing logic
would be cumbersome and inefficient. Hence, we override the command
execution logic that is called by all commands.
)ZHDELZHGETZHLENZHSETZLLENZLPUSHZPUBLISHZRPUSHZRPOPZSADDZSREMZSETZSMEMBERSZZADDZZREMZZREVRANGEBYSCOREr N)�
args_start�args_end���� r )ZDEL�BRPOPZEVALSHAZWATCHc s� t |�}|�d�}|� jv r4� jt|d � |d< nx|� jv r�� j| d }� j| d }|dkrn|d |� ng }g }|d ur�||d � }|� fdd�|||� D � | }|g|�S )Nr rK rL c s g | ]}� j t|� �qS r2 ��global_keyprefix�str��.0�arg��selfr2 r3 �
<listcomp>� s �z5GlobalKeyPrefixMixin._prefix_args.<locals>.<listcomp>)�list�pop�PREFIXED_SIMPLE_COMMANDSrQ rR �PREFIXED_COMPLEX_COMMANDS)rW �args�commandrK rL Zpre_argsZ post_argsr2 rV r3 � _prefix_args� s"
��z!GlobalKeyPrefixMixin._prefix_argsc sH t � j||fi |��}|dkrD|rD|\}}|t| j�d� }||fS |S )z�Parse a response from the Redis server.
Method wraps ``redis.parse_response()`` to remove prefixes of keys
returned by redis command.
rO N)�super�parse_response�lenrQ )rW �
connection� command_name�options�ret�key�value�� __class__r2 r3 ra � s z#GlobalKeyPrefixMixin.parse_responsec s t � j| �|�i |��S rE �r` �execute_commandr_ �rW r] �kwargsri r2 r3 rl s z$GlobalKeyPrefixMixin.execute_commandTc C s t | j| j||| jd�S )N�rQ )�PrefixedRedisPipeline�connection_poolZresponse_callbacksrQ )rW � transactionZ
shard_hintr2 r2 r3 �pipeline s �z GlobalKeyPrefixMixin.pipeline)TN) r8 r9 r: r; r[ r\ r_ ra rl rs �
__classcell__r2 r2 ri r3 rJ � s �
rJ c @ s e Zd ZdZdd� Zdd� ZdS )�PrefixedStrictRedisz@Returns a ``StrictRedis`` client that prefixes the keys it uses.c O s, |� dd�| _tjj| g|�R i |�� d S �NrQ � )rZ rQ r �Redis�__init__rm r2 r2 r3 ry s z PrefixedStrictRedis.__init__c K s t | jfd| ji|��S )NrQ )�PrefixedRedisPubSubrq rQ )rW rn r2 r2 r3 �pubsub s ���zPrefixedStrictRedis.pubsubN)r8 r9 r: r; ry r{ r2 r2 r2 r3 ru s ru c @ s e Zd ZdZdd� ZdS )rp a Custom Redis pipeline that takes global_keyprefix into consideration.
As the ``PrefixedStrictRedis`` client uses the `global_keyprefix` to prefix
the keys it uses, the pipeline called by the client must be able to prefix
the keys as well.
c O s. |� dd�| _tjjj| g|�R i |�� d S rv )rZ rQ r rB �Pipelinery rm r2 r2 r3 ry & s z PrefixedRedisPipeline.__init__N)r8 r9 r: r; ry r2 r2 r2 r3 rp s rp c sD e Zd ZdZdZ� fdd�Zdd� Z� fdd�Z� fd d
�Z� Z S ) rz zCRedis pubsub client that takes global_keyprefix into consideration.)Z SUBSCRIBEZ UNSUBSCRIBEZ
PSUBSCRIBEZ PUNSUBSCRIBEc s$ |� dd�| _t� j|i |�� d S rv )rZ rQ r` ry rm ri r2 r3 ry 5 s z PrefixedRedisPubSub.__init__c s8 t |�}|�d�}|� jv r.� fdd�|D �}|g|�S )Nr c s g | ]}� j t|� �qS r2 rP rS rV r2 r3 rX > s �z4PrefixedRedisPubSub._prefix_args.<locals>.<listcomp>)rY rZ �PUBSUB_COMMANDS)rW r] r^ r2 rV r3 r_ 9 s
�z PrefixedRedisPubSub._prefix_argsc sF t � j|i |��}|du r |S |�^}}}|g� fdd�|D ��|�S )z�Parse a response from the Redis server.
Method wraps ``PubSub.parse_response()`` to remove prefixes of keys
returned by redis command.
Nc s g | ]}|t � j�d � �qS rE )rb rQ )rT rH rV r2 r3 rX W � z6PrefixedRedisPubSub.parse_response.<locals>.<listcomp>)r` ra )rW r] rn rf Z message_typeZchannels�messageri rV r3 ra E s ���z"PrefixedRedisPubSub.parse_responsec s t � j| �|�i |��S rE rk rm ri r2 r3 rl [ s z#PrefixedRedisPubSub.execute_command)
r8 r9 r: r; r} ry r_ ra rl rt r2 r2 ri r3 rz + s rz c s� e Zd ZdZdZ� fdd�Z� fdd�Zd#dd �Z� fd
d �Zd$� fd
d� Z e
d%dd��Z d&dd�Z d'dd�Z
d(dd�Zedd� �Zedd � �Zed d � �Zed d � �Zed!d"� �Z� ZS ))�QoSzRedis Ack Emulation.Tc s t � j|i |�� d| _d S )Nr )r` ry �_vrestore_countrm ri r2 r3 ry d s z QoS.__init__c s� |j }|d |d }}tjd dkr4|t� ig}n
t� |g}| �� �N}|j| jg|�R � �| j|t |j
||g��� � t � �
||� W d � n1 s�0 Y d S )N�exchange� routing_keyr r )�
delivery_infor �VERSIONr �pipe_or_acquireZzadd�unacked_index_keyZhset� unacked_keyr �_raw�executer` �append)rW r � delivery_tagZdelivery�EX�RKZ zadd_args�piperi r2 r3 r� h s
�z
QoS.appendNc C sT | j �|��*}| jD ]}| j||d� qW d � n1 s<0 Y | j�� d S )N�rB )rH �conn_or_acquireZ
_delivered�restore_by_tag�clear)rW rB �tagr2 r2 r3 �restore_unackedy s
.zQoS.restore_unackedc s | � |��� t� �|� d S rE )�_remove_from_indicesr� r` �ack)rW r� ri r2 r3 r� s zQoS.ackFc s2 |r| j |dd� n| �|��� t� �|� d S �NT)�leftmost)r� r� r� r` r� )rW r� Zrequeueri r2 r3 �reject� s z
QoS.rejectc c sF |r |V n6| j �|��}|�� V W d � n1 s80 Y d S rE )rH r� rs )rW r� rB r2 r2 r3 r� � s zQoS.pipe_or_acquirec C sF | � |��(}|�| j|��| j|�W d � S 1 s80 Y d S rE )r� Zzremr� �hdelr� )rW r� r� r2 r2 r3 r� � s �zQoS._remove_from_indicesr �
c
C s� | j d7 _ | j d | r d S | j�� ��}t� | j }znt|| j| j��J |j| j |d|o^||dd�}|png D ]\}}| �
||� qpW d � n1 s�0 Y W n t y� Y n0 W d � n1 s�0 Y d S )Nr r T)�start�numZ
withscores) r� rH r� r �visibility_timeoutrD �unacked_mutex_key�unacked_mutex_expireZzrevrangebyscorer� r� r6 ) rW r� r� �intervalrB �ceilZvisibler� Zscorer2 r2 r3 �restore_visible� s" �
�0 zQoS.restore_visiblec sN � ��fdd�}�j �|�� }|�|�j� W d � n1 s@0 Y d S )Nc sP | � �j��}| �� ���| � |rLtt|��\}}}�j�|||| � � d S rE )�hgetr� �multir� r r rH �_do_restore_message)r� �p�Mr� r� �r� rW r� r2 r3 �restore_transaction� s z/QoS.restore_by_tag.<locals>.restore_transaction)rH r� rr r� )rW r� rB r� r� r2 r� r3 r� � s zQoS.restore_by_tagc C s | j jS rE )rH r� rV r2 r2 r3 r� � s zQoS.unacked_keyc C s | j jS rE )rH r� rV r2 r2 r3 r� � s zQoS.unacked_index_keyc C s | j jS rE )rH r� rV r2 r2 r3 r� � s zQoS.unacked_mutex_keyc C s | j jS rE )rH r� rV r2 r2 r3 r� � s zQoS.unacked_mutex_expirec C s | j jS rE )rH r� rV r2 r2 r3 r� � s zQoS.visibility_timeout)N)F)NN)N)r r� r� )NF)r8 r9 r: r; Zrestore_at_shutdownry r� r� r� r� r r� r� r� r� r r� r� r� r� r� rt r2 r2 ri r3 r� _ s,
r� c @ s� e Zd ZdZeeB ZdZdZdd� Z dd� Z
dd � Z d
d � Z d d
� Z
dd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� Zd(d$d%�Zed&d'� �ZdS ))�MultiChannelPollerz%Async I/O poller for Redis transport.FNc C s( t � | _i | _i | _t� | _t � | _d S rE )�set� _channels� _fd_to_chan�
_chan_to_sockr �poller�
after_readrV r2 r2 r3 ry � s
zMultiChannelPoller.__init__c
C sZ | j �� D ],}z| j�|� W q
ttfy4 Y q
0 q
| j�� | j�� | j �� d S rE ) r� �valuesr� �
unregister�KeyError�
ValueErrorr� r� r� )rW �fdr2 r2 r3 �close� s
zMultiChannelPoller.closec C s | j �|� d S rE )r� �add�rW rH r2 r2 r3 r� � s zMultiChannelPoller.addc C s | j �|� d S rE )r� �discardr� r2 r2 r3 r� � s zMultiChannelPoller.discardc C s. z| j �|j� W n ttfy( Y n0 d S rE )r� r� �_sock�AttributeError� TypeError�rW rc r2 r2 r3 �_on_connection_disconnect� s z,MultiChannelPoller._on_connection_disconnectc C sr |||f| j v r | �|||� |jjd u r4|j�� |jj}||f| j|�� < || j |||f< | j�|| j � d S rE )
r� � _unregisterrc r� �connectr� �filenor� �register�
eventflags)rW rH rB �type�sockr2 r2 r3 � _register� s
z MultiChannelPoller._registerc C s | j �| j|||f � d S rE )r� r� r� )rW rH rB r� r2 r2 r3 r� s z MultiChannelPoller._unregisterc C sL t |dd �d u r0tr"|j�� |_n|j�d�|_|jjd uoJ|||f| jv S )Nrc �_)�getattr�"_REDIS_GET_CONNECTION_WITHOUT_ARGSrq �get_connectionrc r� r� )rW rH rB �cmdr2 r2 r3 �_client_registered s �z%MultiChannelPoller._client_registeredc C s> ||j df}| �||j d�s,d|_| j|� |js:|�� dS )z Enable BRPOP mode for channel.rO FN)rB r� �_in_pollr� � _brpop_start)rW rH �identr2 r2 r3 �_register_BRPOP
s
z"MultiChannelPoller._register_BRPOPc C s8 | � ||jd�s&d|_| �||jd� |js4|�� dS )z Enable LISTEN mode for channel.�LISTENFN)r� � subclient�
_in_listenr� �
_subscriber� r2 r2 r3 �_register_LISTEN s
z#MultiChannelPoller._register_LISTENc C s: | j D ].}|jr$|j�� r$| �|� |jr| �|� qd S rE )r� �
active_queues�qos� can_consumer� �active_fanout_queuesr� r� r2 r2 r3 �
on_poll_start s
z MultiChannelPoller.on_poll_startc C s( || _ | jD ]}|jj|jd� S d S �N)r� )r� r� r� r� �unacked_restore_limit)rW r� rH r2 r2 r3 � on_poll_init& s
�z MultiChannelPoller.on_poll_initc C s* | j D ] }|jr|jj|jd� S qd S r� )r� r� r� r� r� r� r2 r2 r3 �maybe_restore_messages- s
�z)MultiChannelPoller.maybe_restore_messagesc C s< | j D ]0}|j�d�}|d urtt|dd ��r|�� qd S )Nr� � check_health)r� �__dict__�get�callabler� r� )rW rH rB r2 r2 r3 � maybe_check_subclient_health5 s
�z/MultiChannelPoller.maybe_check_subclient_healthc C s( | j | \}}|j�� r$|j| � d S rE )r� r� r� �handlers)rW r� �chanr� r2 r2 r3 � on_readable= s
z MultiChannelPoller.on_readablec C s: |t @ r| �|�| fS |t@ r6| j| \}}|�|� d S rE )r r� r r� � _poll_error)rW r� �eventr� r� r2 r2 r3 � handle_eventB s
z MultiChannelPoller.handle_eventc C sB d| _ z�| jD ].}|jr,|j�� r,| �|� |jr| �|� q| j� |�}|r�|D ]Z\}}| �
||�}|rR W d| _ | j r�z| j � � }W n t
y� Y q�Y qt0 |� qtd S qR| �� t� �W d| _ | j r�z| j � � }W n t
y� Y q�Y q�0 |� q�nDd| _ | j �r<z| j � � }W n t
�y0 Y �q<Y n0 |� �q 0 d S )NTF)�_in_protected_readr� r� r� r� r� r� r� r� r r� r� rZ r� r� r ) rW �callbackr= rH �eventsr� r� rf Zfunr2 r2 r3 r� I sL
�
� zMultiChannelPoller.getc C s | j S rE )r� rV r2 r2 r3 �fdsg s zMultiChannelPoller.fds)N) r8 r9 r: r; r r r� r� r� ry r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� �propertyr� r2 r2 r2 r3 r� � s.
r� c sz e Zd ZdZeZdZdZdZdZdZ dZ
dZ dZ dZ
i ZdZdZd Zd
Zd ZdZd ZeZdZdZdZdZdZ d
Z e Z dZ dZ!dZ"dZ#dZ$dZ%e&j'j(d Z(e)r�e)j*ndZ+e)r�e)j,ndZ-� fdd�Z.dd� Z/dd� Z0dd� Z1drdd�Z2ds� fdd � Z3d d � Z4� fd d �Z5d!d"� Z6� fd#d$�Z7d%d&� Z8d'd(� Z9d)d*� Z:d+d,� Z;d-d.� Z<d/d0� Z=d1d2� Z>dtd4d5�Z?d6d7� Z@d8d9� ZAd:d;� ZBd<d=� ZCd>d?� ZDd@dA� ZEdBdC� ZFdDdE� ZGdudFdG�ZHdHdI� ZIdJdK� ZJdLdM� ZKdNdO� ZLdPdQ� ZM� fdRdS�ZNdTdU� ZOdVdW� ZPdvdXdY�ZQdwdZd[�ZRdxd\d]�ZSdyd^d_�ZTd`da� ZUeVdzdbdc��ZWeXddde� �ZYeXdfdg� �ZZe[dhdi� �Z\e[djdk� �Z]dldm� Z^dndo� Z_eXdpdq� �Z`� ZaS ){�ChannelzRedis Channel.NFTz_kombu.binding.%sz/{db}.zZunackedZ
unacked_indexZ
unacked_mutexi, i r� rw Z round_robin)�sep�
ack_emulationr� r� r� r� r� r� �
fanout_prefix�fanout_patternsrQ �socket_timeout�socket_connect_timeout�socket_keepalive�socket_keepalive_options�queue_order_strategy�max_connections�health_check_interval�retry_on_timeout�priority_stepsc s� t � j|i |�� | js tj| _d| _t| j�� | _| � � | _
| � � | _ t
� | _t
� | _i | _| j| jd�| _| jr�t| jt�r�| j| _nd| _z| j�� W n ty� | �� � Y n0 | j j � | � d| _| j j | _ t d ur�t | t!� d S )NF)rO r� rw T)"r` ry r� r r� � _registeredr r� � _queue_cycle� _get_client�Client�_get_response_errorr1 r� r� �auto_delete_queues�_fanout_to_queue� _brpop_read�_receiver� r� �
isinstancerR �keyprefix_fanoutrB Zping� Exception�_disconnect_poolsrc �cycler� r" r rI rm ri r2 r3 ry � s2
zChannel.__init__c C s | � � d S rE )r rV r2 r2 r3 rF � s zChannel._after_forkc C s< | j }| j}d | _| _ |d ur(|�� |d ur8|�� d S rE )�_pool� _async_pool�
disconnect)rW �pool�
async_poolr2 r2 r3 r s zChannel._disconnect_poolsc C s@ | j |u rd | _ | j|u r d | _| jr<| jjr<| jj�|� d S rE )r� r� rc r r� r� r2 r2 r3 r� s
z!Channel._on_connection_disconnectc C s� zzz d|d d< d|d d d< W n t y4 Y n0 | �||�D ]4}| j|dd�}|r^|jn|j| �||�t|�� qBW n ty� td|dd � Y n0 d S )
NT�headersZ redeliveredZ
propertiesr� F��reversez Could not restore message: %r��exc_info) r� Z_lookup�_get_message_priority�lpushZrpush�
_q_for_prir r
�crit)rW �payloadr� r� r� r� �queue�prir2 r2 r3 r� s �
zChannel._do_restore_messagec sb �j st� �|�S |j�� ��fdd�}��� � }|�|�j� W d � n1 sT0 Y d S )Nc sP | � �j��}| �� | ��j�� |rLtt|��\}}}��|||| � � d S rE )r� r� r� r� r r r� )r� �Pr� r� r� r� r2 r3 r� , s z-Channel._restore.<locals>.restore_transaction)r� r` �_restorer� r� rr r� )rW r r� r� rB ri r� r3 r ' s
zChannel._restorec C s | j |dd�S r� )r )rW r r2 r2 r3 �_restore_at_beginning7 s z Channel._restore_at_beginningc sT || j v r.| j | \}}| j�|� || j|< t� j|g|�R i |��}| �� |S rE )�_fanout_queuesr� r� r r` �
basic_consume�_update_queue_cycle)rW r r] rn r� r� rf ri r2 r3 r" : s
zChannel.basic_consumec C s8 | j }|r4|jjr*|jj�t| j|f��S | �|�S d S rE )rc r r� r� r� r
�
_basic_cancel)rW � consumer_tagrc r2 r2 r3 � basic_cancelN s �zChannel.basic_cancelc s� z| j | }W n ty" Y d S 0 z| j�|� W n tyF Y n 0 | �|� z | j| \}}| j�|� W n ty� Y n0 t� � |�}| �
� |S rE ) Z
_tag_to_queuer� r� �remove�_unsubscribe_fromr! r rZ r` r& r# )rW r% r r� r� rf ri r2 r3 r$ [ s"
zChannel._basic_cancelc C s. |r | j r d�| j|d|g�S d�| j|g�S )Nrw �/)r� �joinr )rW r� r� r2 r2 r3 �_get_publish_topico s
zChannel._get_publish_topicc C s | j | \}}| �||�S rE )r! r+ )rW r r� r� r2 r2 r3 �_get_subscribe_topict s z Channel._get_subscribe_topicc sN � fdd�� j D �}|s d S � j}|jjd u r8|j�� |j� _|�|� d S )Nc s g | ]}� � |��qS r2 )r, �rT r rV r2 r3 rX y s �z&Channel._subscribe.<locals>.<listcomp>)r� r� rc r� r� r� Z
psubscribe)rW �keys�cr2 rV r3 r� x s
�
zChannel._subscribec C s. | � |�}| j}|jr*|jjr*|�|g� d S rE )r, r� rc r� � unsubscribe)rW r �topicr/ r2 r2 r3 r( � s
zChannel._unsubscribe_fromc C s� t |d �dkr&|d dkr&d|_d S t |d �dkr\|d |d |d |d f\}}}}n |d d |d |d f\}}}}||||d�S ) Nr r0 rN FZpmessager r )r� �patternrH �data)r Z
subscribed)rW rB �rr� r2 rH r3 r2 r2 r3 �_handle_message� s & �zChannel._handle_messagec C sd | j }g }z|�| �|�� W n ty0 Y n0 |jd ur\|jjdd�r\|�| �|�� q2t|�S )Nr r< )r� r� � _receive_oner rc Zcan_read�any)rW r/ rf r2 r2 r3 r � s zChannel._receivec C s
d }z |� � }W n | jy, d | _� Y n0 t|ttf��r| �||�}t|d ��d��rt|d �}|d �r|d dkr�|� d�\}}}zt
t|d ��}W n8 t t fy� t
d|t|�d d � d
d � t� �Y n0 |�dd
�d }| j�|| j| � d S d S )
Nr� r rH r3 r r) �.z&Cannot process event on channel %r: %si r r T)ra r" r� r rY �tupler5 r �endswith� partitionr r� r� �warning�reprr �splitrc �_deliverr )rW r/ �responser rH r� r r� r2 r2 r3 r6 � s2
�
�zChannel._receive_oner c sr �j �t�j��� � sd S � �fdd��jD �|p4dg }�jj�_dg|�}�jr`�j� |�}�jjj
|� d S )Nc s" g | ]}� D ]}�� ||��q qS r2 )r )rT r r ��queuesrW r2 r3 rX � s �z(Channel._brpop_start.<locals>.<listcomp>r rO ) r Zconsumerb r� r� rB rc r� rQ r_ Z send_command)rW r= r. � command_argsr2 rA r3 r� � s �
zChannel._brpop_startc K s� z�z | j j| j jdfi |��}W n" | jyB | j j�� � Y n0 |r�|\}}t|��| jd�d }| j� |� | j�
t t|��|� W d | _ dS t
� �W d | _ nd | _ 0 d S )NrO r r T)rB ra rc r" r r �rsplitr� r �rotater? r r� r )rW re Z
dest__item�dest�itemr2 r2 r3 r � s( ��
�zChannel._brpop_readc K s* |dkr| j �� n| j�| jj|� d S )Nr� )r� ra rB rc )rW r� re r2 r2 r3 r� � s zChannel._poll_errorc C sr | � � �V}| jD ]8}|�| �||��}|rtt|�� W d � S qt� �W d � n1 sd0 Y d S rE )r� r� Zrpopr r r r )rW r rB r rG r2 r2 r3 �_get� s
z Channel._getc C s� | � � �~}|�� �V}| jD ]}|�| �||��}q|�� }tdd� |D ��W d � W d � S 1 sn0 Y W d � n1 s�0 Y d S )Nc s s | ]}t |tj�r|V qd S rE )r �numbers�Integral)rT �sizer2 r2 r3 � <genexpr>� s �z Channel._size.<locals>.<genexpr>)r� rs r� �llenr r� �sum)rW r rB r� r �sizesr2 r2 r3 �_size� s
z
Channel._sizec C s$ | � |�}|r |� | j� |� �S |S rE )�priorityr� )rW r r r2 r2 r3 r � s
zChannel._q_for_pric C s | j }|t||�d S )Nr )r� r )rW �nZstepsr2 r2 r3 rQ � s zChannel.priorityc K sR | j |dd�}| �� �(}|�| �||�t|�� W d � n1 sD0 Y dS )zDeliver message.Fr N)r r� r r r )rW r r rn r rB r2 r2 r3 �_put s
z Channel._putc K sD | � � �(}|�| �||�t|�� W d � n1 s60 Y dS )zDeliver fanout message.N)r� �publishr+ r )rW r� r r� rn rB r2 r2 r3 � _put_fanout s
�zChannel._put_fanoutc K s |r| j �|� d S rE )r r� )rW r Z auto_deletern r2 r2 r3 �
_new_queue s zChannel._new_queuec C s� | � |�jdkr&||�dd�f| j|< | �� �>}|�| j|f | j�|pJd|pPd|pVdg�� W d � n1 sr0 Y d S )N�fanout�#�*rw ) Ztypeofr� �replacer! r� Zsadd�keyprefix_queuer� r* )rW r� r� r2 r rB r2 r2 r3 � _queue_bind s �
��zChannel._queue_bindc
O s� | j �|� | j|�d�d���}|�| j|f | j�|p:d|p@d|pFdg�� |�� �6}| j D ]} |�
| � || ��}q^|� � W d � n1 s�0 Y W d � n1 s�0 Y d S )NrB r� rw )
r r� r� r� Zsremr[ r� r* rs r� �deleter r� )
rW r r� r� r2 r] rn rB r� r r2 r2 r3 �_delete! s ��
zChannel._deletec K s� | � � �p}|�� �H}| jD ]}|�| �||��}qt|�� �W d � W d � S 1 s`0 Y W d � n1 s~0 Y d S rE )r� rs r� �existsr r7 r� )rW r rn rB r� r r2 r2 r3 �
_has_queue- s
zChannel._has_queuec sh � j | }� �� �B}|�|�}|s4g W d � S � fdd�|D �W d � S 1 sZ0 Y d S )Nc s g | ]}t t|��� j���qS r2 )r9 r r> r� )rT �valrV r2 r3 rX < r~ z%Channel.get_table.<locals>.<listcomp>)r[ r� Zsmembers)rW r� rg rB r� r2 rV r3 � get_table4 s
zChannel.get_tablec C s� | � � ��}|�� �`}| jD ] }| �||�}|�|��|�}q|�� }t|d d d� �W d � W d � S 1 sx0 Y W d � n1 s�0 Y d S )NrN )r� rs r� r rM r] r� rN )rW r rB r� r ZpriqrO r2 r2 r3 �_purge> s
zChannel._purgec s� d| _ | jr,z | �� W n ty* Y n0 | js�| jj�| � | j� d�}|d urx| j
D ] }|| j v rZ| j ||d� qZ| �
� | �� t� �� d S )NTrB r� )�_closingr� r r �closedrc r r� r� r� r! r Z queue_deleter �_close_clientsr` r� )rW rB r ri r2 r3 r� G s
z
Channel.closec C sN dD ]D}z$| j | }|jd }|_|�� W q tt| jfyF Y q0 qd S )N)rB r� )r� rc r r� r� r1 )rW �attrrB rc r2 r2 r3 rf \ s
zChannel._close_clientsc C sf t |tj�sb|r|dkr t}n|�d�r4|dd � }z t|�}W n ty` td�|���Y n0 |S )Nr) r z/Database is int between 0 and limit - 1, not {})r rI rJ �
DEFAULT_DB�
startswith�intr� �format)rW Zvhostr2 r2 r3 �_prepare_virtual_hostf s
��
z Channel._prepare_virtual_hostc K s |S rE r2 )rW r� r� �paramsr2 r2 r3 �_filter_tcp_connparamsu s z Channel._filter_tcp_connparamsc
s� | j j}|jpd|jp | j j|j|j|j| j| j | j
| j | j | j
| jd� }| j}t|d�r�|g}t|d�rx|t|j�7 }|D ]}t|jd�r| q�q||�d� |jr�z|�|j� | j|d< W n ty� Y n0 |d }d|v �r|t|�\}} } }
} } }
|d k�rT| jf i |��}|jt j d
| d �fi |
�� |�d d � |�d
d � |�dd � |
|d< | |d< |�dd � |�dd � | � |�dd ��|d< | � |� d��p�| j}|�r�G � fdd�d|�}|}||d< |S )Nz 127.0.0.1) �host�port� virtual_host�username�passwordr� r� r� r� r� r� r� ry � __bases__r� �connection_classro z://r* r) )ru �pathr� r� r� rr rs rp rq �dbc s e Zd Z� �fdd�Z� ZS )z'Channel._connparams.<locals>.Connectionc s t � j|� �jr ��| � d S rE )r` r r� r� )rW r] )rj rH r2 r3 r � s z2Channel._connparams.<locals>.Connection.disconnect)r8 r9 r: r rt r2 rG ri r3 �
Connection� s rx ) rc rB �hostnamerp � default_portrq Zuseridrs r� r� r� r� r� r� r� ru r' rY rt r ry rZ �ssl�update�connection_class_sslr� r rn r ZUnixDomainSocketConnectionrl r� )rW � asynchronousZconninfo�
connparamsZ
conn_class�classes�klassro �schemer� rr rs rv �queryZconnection_clsrx r2 rG r3 � _connparamsy sv �
��
� �zChannel._connparamsc C s |r| j | jd�S | j | jd�S )N)rq )r r r )rW r~ r2 r2 r3 �_create_client� s zChannel._create_clientc C s0 | j |d�}| jj|d d�| _tjf i |��S �N�r~ rw )rw )r� r rk r �ConnectionPool�rW r~ rm r2 r2 r3 � _get_pool� s zChannel._get_poolc C s4 t jdk rtd�t ���| jr.tjt| jd�S t jS )N)r rN r zSRedis transport requires redis-py versions 3.2.0 or later. You have {0.__version__}ro ) r r� r rk rQ � functools�partialru rx rV r2 r2 r3 r � s
���zChannel._get_clientc c s |r |V n
| � � V d S rE �r� �rW rB r2 r2 r3 r� � s zChannel.conn_or_acquirec C s | j d u r| �� | _ | j S rE )r
r� rV r2 r2 r3 r � s
z Channel.poolc C s | j d u r| jdd�| _ | j S )NTr� )r r� rV r2 r2 r3 r � s
zChannel.async_poolc C s | j dd�S )z+Client used to publish messages, BRPOP etc.Tr� r� rV r2 r2 r3 rB � s zChannel.clientc C s | j dd�}|�� S )z1Pub/Sub connection used to consume fanout queues.Tr� )r� r{ r� r2 r2 r3 r� � s zChannel.subclientc C s | j �| j� d S rE )r r| r� rV r2 r2 r3 r# s zChannel._update_queue_cyclec C s ddl m} |jS )Nr r$ )r r% r1 )rW r% r2 r2 r3 r s zChannel._get_response_errorc s � fdd�� j D �S )z<Set of queues being consumed from (excluding fanout queues).c s h | ]}|� j vr|�qS r2 )r� r- rV r2 r3 � <setcomp>
s
�z(Channel.active_queues.<locals>.<setcomp>)Z_active_queuesrV r2 rV r3 r� s zChannel.active_queues)F)F)r )F)NN)F)F)F)N)br8 r9 r: r; r� Z_clientZ
_subclientrd Zsupports_fanoutr[ r r� r� r� r! r� r� r� r� r� r� r� �PRIORITY_STEPSr� r� r� r� r� r� r� � DEFAULT_HEALTH_CHECK_INTERVALr� r� r� rQ r� r r
r r� �from_transport_optionsr rx ru �
SSLConnectionr} ry rF r r� r� r r r" r& r$ r+ r, r� r( r5 r r6 r� r r� rH rP r rQ rS rU rV r\ r^ r` rb rc r� rf rl rn r� r� r� r r r� r� r r r rB r� r# r r� rt r2 r2 ri r3 r� l s� �� ( �
�
Q
r� c st e Zd ZdZeZdZeZdZdZ e
j j j
deg d��d�Z erHe� \ZZ� fdd�Zd d
� Zd d � Zd
d� Z� ZS )r) zRedis Transport.Nr T)�directr1 rW )r~ Z
exchange_typec s. t d u rtd��t� j|i |�� t� | _d S )Nz)Missing redis library (pip install redis))r � ImportErrorr` ry r� r rm ri r2 r3 ry s zTransport.__init__c C s t jS rE )r � __version__rV r2 r2 r3 �driver_version( s zTransport.driver_versionc s� | j ����j� �j��j� | j����fdd�}|�_� ���fdd���j��� �� d�j
� |j j �
dt�}�� |�j� d S )Nc s@ | j r��| j � � jr<z�j��� W n ty: Y n0 d S rE )r� r' r� �on_tickr� )rc )r �loopr� r2 r3 �_on_disconnect2 s z:Transport.register_with_event_loop.<locals>._on_disconnectc s �� � �fdd��j D � d S )Nc s g | ]}� |�|��qS r2 r2 )rT r� )�
add_readerr� r2 r3 rX A r~ zMTransport.register_with_event_loop.<locals>.on_poll_start.<locals>.<listcomp>)r� r2 )r� r �cycle_poll_startr� r2 r3 r� ? s z9Transport.register_with_event_loop.<locals>.on_poll_startr� r� )r r� r� r� r� r� r� r� r� Zcall_repeatedlyr� rB Ztransport_optionsr� r� r� )rW rc r� r� r� r2 )r� r r� r� r� r� r3 �register_with_event_loop+ s$ ��z"Transport.register_with_event_loopc C s | j �|� dS )z1Handle AIO event for one of our file descriptors.N)r r� )rW r� r2 r2 r3 r� M s zTransport.on_readable)r8 r9 r: r; r� Zpolling_interval� DEFAULT_PORTrz Z driver_typeZ driver_namer r) Z
implements�extend� frozensetr r4 r" r# ry r� r� r� rt r2 r2 ri r3 r) s
�
"r) c @ s e Zd ZdZdS )� SentinelManagedSSLConnectionz�Connect to a Redis server using Sentinel + TLS.
Use Sentinel to identify which Redis server is the current master
to connect to and when connecting to the Master server, use an
SSL Connection.
Nr7 r2 r2 r2 r3 r� S s r� c @ sH e Zd ZdZejd Zer ejndZer,e ndZ
d dd�Z d
dd�Z dS ) �SentinelChannela Channel with explicit Redis Sentinel knowledge.
Broker url is supposed to look like:
.. code-block::
sentinel://0.0.0.0:26379;sentinel://0.0.0.0:26380/...
where each sentinel is separated by a `;`.
Other arguments for the sentinel should come from the transport options
(see `transport_options` of :class:`~kombu.connection.Connection`).
You must provide at least one option in Transport options:
* `master_name` - name of the redis group to poll
Example:
-------
.. code-block:: python
>>> import kombu
>>> c = kombu.Connection(
'sentinel://sentinel1:26379;sentinel://sentinel2:26379',
transport_options={'master_name': 'mymaster'}
)
>>> c.connect()
)� master_name�min_other_sentinels�sentinel_kwargsNFc C s� | � |�}|�� }|�dd � |�dd � g }| jjjD ]4}t|�}|jdkr8|jpZ| jj }|�
|j |f� q8|s�|�
|d |d f� t j
|ft| dd�t| dd �d�|��}t| dd �}|d u r�td ��|�|tj�jS )
Nro rp r r� r r� )r� r� r� z1'master_name' transport option must be specified.)r� �copyrZ rc rB Zaltr r� rp rz r� ry r ZSentinelr� r� Z
master_forr rx rq ) rW r~ r Zadditional_paramsZ sentinels�urlrp Z
sentinel_instr� r2 r2 r3 �_sentinel_managed_pool� s:
�
�� ��z&SentinelChannel._sentinel_managed_poolc C s* | j |d�}| jj|d d�| _| �|�S r� )r� r rk r� r� r2 r2 r3 r� � s zSentinelChannel._get_pool)F)F)
r8 r9 r: r; r� r� r �SentinelManagedConnectionru r� r} r� r� r2 r2 r2 r3 r� ` s
%r� c @ s e Zd ZdZdZeZdS )�SentinelTransportzRedis Sentinel Transport.i g N)r8 r9 r: r; rz r� r� r2 r2 r2 r3 r� � s r� )Rr; �
__future__r r� rI r* r � collectionsr �
contextlibr �importlib.metadatar r r r Zpackaging.versionr Zviner
Zkombu.exceptionsr r Z kombu.logr
Zkombu.utils.compatr Zkombu.utils.encodingr Zkombu.utils.eventior r r Zkombu.utils.functionalr Zkombu.utils.jsonr r Zkombu.utils.objectsr Zkombu.utils.schedulingr Zkombu.utils.urlr rw r r r� r� r �logger�criticalr< r r� rh r� r� r! r4 r5 r
r6 rD rI rJ rx ru rB r| rp ZPubSubrz r� r� r� r) r� r� r� r� r� r2 r2 r2 r3 �<module> s� 5
S
4k # 'D
�
P