o
    wvXjö�  ã                   @   s6  d Z ddlmZmZ ddl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mZ dd
lmZmZmZ ddlmZ ddlmZ ddlmZmZ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dl)m*Z* ddl+m,Z, zddl-Z-W n e.y›   dZ-Y nw zddl-m/Z/ W n e.y¯   dZ/Y nw edƒZ0e0j1e0j2Z3Z2dZ4dZ5dZ6g d¢Z7eddƒZ8dZ9dd „ Z:d!d"„ Z;G d#d$„ d$e<ƒZ=e
d%d&„ ƒZ>d'd(„ Z?G d)d*„ d*e,j@ƒZ@G d+d,„ d,eAƒZBG d-d.„ d.e,jCƒZCG d/d0„ d0e,jDƒZDG d1d2„ d2eCƒZEG d3d4„ d4eDƒZFdS )5zRedis transport.é    )Úabsolute_importÚunicode_literalsN)Úbisect)Ú
namedtuple)Úcontextmanager)Útime)Úpromise)ÚInconsistencyErrorÚVersionMismatch)ÚEmptyÚvaluesÚstring_t)Ú
get_logger)Úregister_after_fork)ÚpollÚREADÚERR)Úbytes_to_str)ÚloadsÚdumps)Úcached_property)Úcycle_by_name)Ú
_parse_url)Ú_detect_environment)Úaccepts_argumenté   )Úvirtual)Úsentinelzkombu.transport.redisië  é   )r   é   é   é	   Úerror_classes_t)Úconnection_errorsÚchannel_errorsz’
Cannot route message for exchange {0!r}: Table empty or key no longer exists.
Probably the key ({1!r}) has been removed from the Redis database.
c               	   C   s^   ddl m}  t| dƒr| j}n| j}ttjjt	t
jtt| j| j| jf tjj|| j| jf ƒS )z$Return tuple of redis error classes.r   ©Ú
exceptionsÚInvalidData)Úredisr&   Úhasattrr'   Ú	DataErrorr"   r   Ú	Transportr#   r	   ÚsocketÚerrorÚIOErrorÚOSErrorÚConnectionErrorÚAuthenticationErrorÚTimeoutErrorr$   ÚInvalidResponseÚResponseError)r&   r*   © r5   úR/var/www/html/myproject/venv/lib/python3.10/site-packages/kombu/transport/redis.pyÚget_redis_error_classesJ   s(   
ùý÷r7   c                  C   s   ddl m}  | jS )z1Return the redis ConnectionError exception class.r   r%   )r(   r&   r0   r%   r5   r5   r6   Úget_redis_ConnectionErrorb   s   r8   c                   @   s   e Zd ZdZdS )Ú	MutexHeldz)Raised when another party holds the lock.N)Ú__name__Ú
__module__Ú__qualname__Ú__doc__r5   r5   r5   r6   r9   h   s    r9   c                 c   s�   � | j ||d�}d}z(|jdd�}|rdV  ntƒ ‚W |r1z| ¡  W dS  tjjy0   Y dS w dS |rGz| ¡  W w  tjjyF   Y w w w )zTAcquire redis lock in non blocking way.

    Raise MutexHeld if not successful.
    ©ÚtimeoutF)ÚblockingN)ÚlockÚacquirer9   Úreleaser(   r&   ÚLockNotOwnedError)ÚclientÚnameÚexpirerA   Úlock_acquiredr5   r5   r6   ÚMutexl   s,   €þþýþýrI   c                 C   ó   |   ¡  d S ©N)Ú_after_fork)Úchannelr5   r5   r6   Ú_after_fork_cleanup_channelƒ   ó   rN   c                       s¸   e Zd ZdZdZ‡ fdd„Z‡ fdd„Zd#dd	„Z‡ fd
d„Zd$dd„Z	e
d%dd„ƒZd#dd„Zd&dd„Zd'dd„Zedd„ ƒZedd„ ƒZedd„ ƒZedd „ ƒZed!d"„ ƒZ‡  ZS )(ÚQoSzRedis Ack Emulation.Tc                    s    t t| ƒj|i |¤Ž d| _d S )Nr   )ÚsuperrP   Ú__init__Ú_vrestore_count©ÚselfÚargsÚkwargs©Ú	__class__r5   r6   rR   Œ   s   
zQoS.__init__c              	      s®   |j }|d |d }}tjd dkr|tƒ ig}ntƒ |g}|  ¡ �*}|j| jg|¢R Ž  | j|t	|j
||gƒ¡ ¡  tt| ƒ ||¡ W d   ƒ d S 1 sPw   Y  d S )NÚexchangeÚrouting_keyr   r   )Údelivery_infor(   ÚVERSIONr   Úpipe_or_acquireÚzaddÚunacked_index_keyÚhsetÚunacked_keyr   Ú_rawÚexecuterQ   rP   Úappend)rU   ÚmessageÚdelivery_tagÚdeliveryÚEXÚRKÚ	zadd_argsÚpiperX   r5   r6   re   �   s   

ÿ"ûz
QoS.appendNc                 C   sT   | j  |¡�}| jD ]	}| j||d� q
W d   ƒ n1 sw   Y  | j ¡  d S )N©rE   )rM   Úconn_or_acquireÚ
_deliveredÚrestore_by_tagÚclear)rU   rE   Útagr5   r5   r6   Úrestore_unacked¡   s   
ÿÿzQoS.restore_unackedc                    s"   |   |¡ ¡  tt| ƒ |¡ d S rK   )Ú_remove_from_indicesrd   rQ   rP   Úack)rU   rg   rX   r5   r6   ru   §   s   zQoS.ackFc                 C   s    |r	| j |dd� |  |¡ d S ©NT)Úleftmost)rp   ru   )rU   rg   Úrequeuer5   r5   r6   Úreject«   ó   z
QoS.rejectc                 c   sL   � |r|V  d S | j  |¡�}| ¡ V  W d   ƒ d S 1 sw   Y  d S rK   )rM   rn   Úpipeline)rU   rl   rE   r5   r5   r6   r^   °   s   €
"ÿzQoS.pipe_or_acquirec                 C   sF   |   |¡�}| | j|¡ | j|¡W  d   ƒ S 1 sw   Y  d S rK   )r^   Úzremr`   Úhdelrb   )rU   rg   rl   r5   r5   r6   rt   ¸   s   
$þzQoS._remove_from_indicesr   é
   c           
   
   C   s  |  j d7  _ | j d | rd S | j ¡ �b}tƒ | j }zAt|| j| jƒ�/ tƒ }|dkr0tƒ }|j	| j
|d|o9||dd�}|pAg D ]
\}}	|  ||¡ qBW d   ƒ n1 sWw   Y  W n	 tyf   Y n	w W d   ƒ d S W d   ƒ d S 1 szw   Y  d S )Nr   Úgeventr   T)ÚstartÚnumÚ
withscores)rS   rM   rn   r   Úvisibility_timeoutrI   Úunacked_mutex_keyÚunacked_mutex_expirer   Úzrevrangebyscorer`   rp   r9   )
rU   r€   r�   ÚintervalrE   ÚceilÚenvÚvisiblerr   Úscorer5   r5   r6   Úrestore_visible½   s8   ÿ
þÿø€
ÿ€ó"òzQoS.restore_visiblec           
   	   C   s´   | j  |¡�J}| ¡ �}|  || | j|¡¡ ¡ \}}}W d   ƒ n1 s&w   Y  |rHtt|ƒƒ\}}}	| j  	|||	||¡ W d   ƒ d S W d   ƒ d S 1 sSw   Y  d S rK   )
rM   rn   r{   rt   Úhgetrb   rd   r   r   Ú_do_restore_message)
rU   rr   rE   rw   rl   ÚpÚ_ÚMri   rj   r5   r5   r6   rp   Ñ   s   
ÿ
ÿÿú"üzQoS.restore_by_tagc                 C   ó   | j jS rK   )rM   rb   ©rU   r5   r5   r6   rb   Ú   ó   zQoS.unacked_keyc                 C   r’   rK   )rM   r`   r“   r5   r5   r6   r`   Þ   r”   zQoS.unacked_index_keyc                 C   r’   rK   )rM   r„   r“   r5   r5   r6   r„   â   r”   zQoS.unacked_mutex_keyc                 C   r’   rK   )rM   r…   r“   r5   r5   r6   r…   æ   r”   zQoS.unacked_mutex_expirec                 C   r’   rK   )rM   rƒ   r“   r5   r5   r6   rƒ   ê   r”   zQoS.visibility_timeoutrK   ©F©NN)r   r~   r~   ©NF)r:   r;   r<   r=   Úrestore_at_shutdownrR   re   rs   ru   ry   r   r^   rt   rŒ   rp   r   rb   r`   r„   r…   rƒ   Ú__classcell__r5   r5   rX   r6   rP   ‡   s.    




	



rP   c                   @   s¶   e Zd ZdZeeB ZdZdZdd„ Z	dd„ Z
dd	„ Zd
d„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zd d!„ Zd"d#„ Zd(d$d%„Zed&d'„ ƒZdS ))ÚMultiChannelPollerz%Async I/O poller for Redis transport.FNc                 C   s(   t ƒ | _i | _i | _tƒ | _t ƒ | _d S rK   )ÚsetÚ	_channelsÚ_fd_to_chanÚ_chan_to_sockr   ÚpollerÚ
after_readr“   r5   r5   r6   rR   ú   s
   zMultiChannelPoller.__init__c              
   C   sX   t | jƒD ]}z| j |¡ W q ttfy   Y qw | j ¡  | j ¡  | j ¡  d S rK   )	r   rž   rŸ   Ú
unregisterÚKeyErrorÚ
ValueErrorrœ   rq   r�   )rU   Úfdr5   r5   r6   Úclose  s   ÿ

zMultiChannelPoller.closec                 C   ó   | j  |¡ d S rK   )rœ   Úadd©rU   rM   r5   r5   r6   r§     ó   zMultiChannelPoller.addc                 C   r¦   rK   )rœ   Údiscardr¨   r5   r5   r6   rª     r©   zMultiChannelPoller.discardc              	   C   s.   z
| j  |j¡ W d S  ttfy   Y d S w rK   )rŸ   r¡   Ú_sockÚAttributeErrorÚ	TypeError©rU   Ú
connectionr5   r5   r6   Ú_on_connection_disconnect  s
   ÿz,MultiChannelPoller._on_connection_disconnectc                 C   sr   |||f| j v r|  |||¡ |jjd u r|j ¡  |jj}||f| j| ¡ < || j |||f< | j || j	¡ d S rK   )
rž   Ú_unregisterr¯   r«   Úconnectr�   ÚfilenorŸ   ÚregisterÚ
eventflags)rU   rM   rE   ÚtypeÚsockr5   r5   r6   Ú	_register  s   
zMultiChannelPoller._registerc                 C   s   | j  | j|||f ¡ d S rK   )rŸ   r¡   rž   )rU   rM   rE   r¶   r5   r5   r6   r±   &  s   zMultiChannelPoller._unregisterc                 C   s:   t |dd ƒd u r|j d¡|_|jjd uo|||f| jv S )Nr¯   r�   )ÚgetattrÚconnection_poolÚget_connectionr¯   r«   rž   )rU   rM   rE   Úcmdr5   r5   r6   Ú_client_registered)  s
   ÿz%MultiChannelPoller._client_registeredc                 C   sB   ||j df}|  ||j d¡sd|_| j|Ž  |js| ¡  dS dS )zEnable BRPOP mode for channel.ÚBRPOPFN)rE   r½   Ú_in_pollr¸   Ú_brpop_start)rU   rM   Úidentr5   r5   r6   Ú_register_BRPOP/  s   
ÿz"MultiChannelPoller._register_BRPOPc                 C   s<   |   ||jd¡sd|_|  ||jd¡ |js| ¡  dS dS )zEnable LISTEN mode for channel.ÚLISTENFN)r½   Ú	subclientÚ
_in_listenr¸   Ú
_subscriber¨   r5   r5   r6   Ú_register_LISTEN8  s   ÿz#MultiChannelPoller._register_LISTENc                 C   s:   | j D ]}|jr|j ¡ r|  |¡ |jr|  |¡ qd S rK   )rœ   Úactive_queuesÚqosÚcan_consumerÂ   Úactive_fanout_queuesrÇ   r¨   r5   r5   r6   Úon_poll_start@  s   



€ûz MultiChannelPoller.on_poll_startc                 C   s(   || _ | jD ]}|jj|jd�  S d S ©N)r�   )rŸ   rœ   rÉ   rŒ   Úunacked_restore_limit)rU   rŸ   rM   r5   r5   r6   Úon_poll_initH  s   

ÿÿzMultiChannelPoller.on_poll_initc                 C   s*   | j D ]}|jr|jj|jd�  S qd S rÍ   )rœ   rÈ   rÉ   rŒ   rÎ   r¨   r5   r5   r6   Úmaybe_restore_messagesO  s   

ÿþÿz)MultiChannelPoller.maybe_restore_messagesc                 C   s<   | j D ]}|j d¡}|d urtt|dd ƒƒr| ¡  qd S )NrÄ   Úcheck_health)rœ   Ú__dict__ÚgetÚcallabler¹   rÑ   )rU   rM   rE   r5   r5   r6   Úmaybe_check_subclient_healthW  s   
ÿ€ûz/MultiChannelPoller.maybe_check_subclient_healthc                 C   s,   | j | \}}|j ¡ r|j| ƒ  d S d S rK   )r�   rÉ   rÊ   Úhandlers)rU   r³   Úchanr¶   r5   r5   r6   Úon_readable_  s   
ÿzMultiChannelPoller.on_readablec                 C   s>   |t @ r|  |¡| fS |t@ r| j| \}}| |¡ d S d S rK   )r   rØ   r   r�   Ú_poll_error)rU   r³   Úeventr×   r¶   r5   r5   r6   Úhandle_eventd  s   þzMultiChannelPoller.handle_eventc           	      C   sþ   d| _ z]| jD ]}|jr|j ¡ r|  |¡ |jr|  |¡ q| j 	|¡}|rZ|D ]0\}}|  
||¡}|rY W d| _ | jrWz| j ¡ }W n
 tyN   Y d S w |ƒ  | js=d S d S q)|  ¡  tƒ ‚d| _ | jr~z| j ¡ }W n	 tyw   Y w w |ƒ  | jsgw )NTF)Ú_in_protected_readrœ   rÈ   rÉ   rÊ   rÂ   rË   rÇ   rŸ   r   rÛ   r    Úpopr¢   rÐ   r   )	rU   Úcallbackr?   rM   Úeventsr³   rÚ   ÚretÚfunr5   r5   r6   rÓ   k  sJ   



€ÿúøüúzMultiChannelPoller.getc                 C   s   | j S rK   )r�   r“   r5   r5   r6   Úfds‰  ó   zMultiChannelPoller.fdsrK   )r:   r;   r<   r=   r   r   rµ   rÜ   r    rR   r¥   r§   rª   r°   r¸   r±   r½   rÂ   rÇ   rÌ   rÏ   rÐ   rÕ   rØ   rÛ   rÓ   Úpropertyrâ   r5   r5   r5   r6   rš   ï   s0    

	
rš   c                       sn  e Zd ZdZeZdZdZdZdZdZ	dZ
dZdZdZi ZdZdZd	Zd
ZdZdZdZeZdZdZdZdZdZdZeZdZ dZ!dZ"dZ#dZ$e%j&j'd Z'e(rOe(j)ndZ*‡ fdd„Z+dd„ Z,dd„ Z-dd„ Z.	dqdd„Z/dr‡ fdd„	Z0dd„ Z1‡ fdd„Z2d d!„ Z3‡ fd"d#„Z4d$d%„ Z5d&d'„ Z6d(d)„ Z7d*d+„ Z8d,d-„ Z9d.d/„ Z:d0d1„ Z;dsd3d4„Z<d5d6„ Z=d7d8„ Z>d9d:„ Z?d;d<„ Z@d=d>„ ZAd?d@„ ZBdAdB„ ZCdCdD„ ZDdrdEdF„ZEdGdH„ ZFdIdJ„ ZGdKdL„ ZHdMdN„ ZIdOdP„ ZJ‡ fdQdR„ZKdSdT„ ZLdUdV„ ZM		dtdWdX„ZNdrdYdZ„ZOdrd[d\„ZPdrd]d^„ZQd_d`„ ZReSdudadb„ƒZTeUdcdd„ ƒZVeUdedf„ ƒZWeXdgdh„ ƒZYeXdidj„ ƒZZdkdl„ Z[dmdn„ Z\eUdodp„ ƒZ]‡  Z^S )vÚChannelzRedis Channel.NFTz_kombu.binding.%sz/{db}.zÚunackedÚunacked_indexÚunacked_mutexi,  i  r~   Úround_robin)ÚsepÚack_emulationrb   r`   r„   r…   rƒ   rÎ   Úfanout_prefixÚfanout_patternsÚ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 t| ƒj|i |¤Ž | jstj| _t| jƒƒ | _|  	¡ | _
|  ¡ | _tƒ | _tƒ | _i | _| j| jdœ| _| jrDt| jtƒrC| j| _nd| _z| j ¡  W n tyZ   |  ¡  ‚ w | jj | ¡ | jj| _t d urrt | t!ƒ d S d S )N)r¾   rÃ   Ú )"rQ   rå   rR   rë   r   rP   r   rò   Ú_queue_cycleÚ_get_clientÚClientÚ_get_response_errorr4   r›   rË   Úauto_delete_queuesÚ_fanout_to_queueÚ_brpop_readÚ_receiverÖ   rì   Ú
isinstancer   Úkeyprefix_fanoutrE   ÚpingÚ	ExceptionÚ_disconnect_poolsr¯   Úcycler§   r#   r   rN   rT   rX   r5   r6   rR   ò  s4   

€þ
ÿzChannel.__init__c                 C   rJ   rK   )r  r“   r5   r5   r6   rL     rO   zChannel._after_forkc                 C   s@   | j }| j}d  | _| _ |d ur| ¡  |d ur| ¡  d S d S rK   )Ú_poolÚ_async_poolÚ
disconnect)rU   ÚpoolÚ
async_poolr5   r5   r6   r    s   ÿzChannel._disconnect_poolsc                 C   sH   | j |u rd | _ | j|u rd | _| jr | jjr"| jj |¡ d S d S d S rK   )r¿   rÅ   r¯   r  r°   r®   r5   r5   r6   r°   &  s   

ÿz!Channel._on_connection_disconnectc              	   C   s¶   |   |¡�L}z*zd|d d< W n	 ty   Y nw |  ||¡D ]}|r&|jn|j|t|ƒƒ qW n ty@   td|dd� Y n	w W d   ƒ d S W d   ƒ d S 1 sTw   Y  d S )NTÚheadersÚredeliveredzCould not restore message: %r©Úexc_info)rn   r¢   Ú_lookupÚlpushÚrpushr   r  Úcrit)rU   ÚpayloadrZ   r[   rE   rw   Úqueuer5   r5   r6   rŽ   .  s&   ÿÿÿÿüú"õzChannel._do_restore_messagec              	      sÊ   | j stt| ƒ |¡S |j}|  ¡ �I}| ¡ �}| | j|¡ 	| j|¡ 
¡ \}}W d   ƒ n1 s2w   Y  |rStt|ƒƒ\}}	}
|  ||	|
||¡ W d   ƒ d S W d   ƒ d S 1 s^w   Y  d S rK   )rë   rQ   rå   Ú_restorerg   rn   r{   r�   rb   r}   rd   r   r   rŽ   )rU   rf   rw   rr   rE   rl   ÚPr�   r‘   ri   rj   rX   r5   r6   r  =  s    


þÿù"ûzChannel._restorec                 C   s   | j |dd�S rv   )r  )rU   rf   r5   r5   r6   Ú_restore_at_beginningJ  s   zChannel._restore_at_beginningc                    sX   || j v r| j | \}}| j |¡ || j|< tt| ƒj|g|¢R i |¤Ž}|  ¡  |S rK   )Ú_fanout_queuesrË   r§   rý   rQ   rå   Úbasic_consumeÚ_update_queue_cycle)rU   r  rV   rW   rZ   r�   rà   rX   r5   r6   r  M  s   

zChannel.basic_consumec                 C   s8   | j }|r|jjr|jj t| j|fƒ¡S |  |¡S d S rK   )r¯   r  rÜ   r    r§   r   Ú_basic_cancel)rU   Úconsumer_tagr¯   r5   r5   r6   Úbasic_cancela  s   ÿ
ûzChannel.basic_cancelc                    s    z| j | }W n
 ty   Y d S w z| j |¡ W n	 ty#   Y nw |  |¡ z| j| \}}| j |¡ W n	 tyA   Y nw tt	| ƒ 
|¡}|  ¡  |S rK   )Ú_tag_to_queuer¢   rË   ÚremoveÚ_unsubscribe_fromr  rý   rÝ   rQ   rå   r  r  )rU   r  r  rZ   r�   rà   rX   r5   r6   r  n  s(   ÿÿ
ÿzChannel._basic_cancelc                 C   s.   |r| j rd | j|d|g¡S d | j|g¡S )Nr÷   ú/)rí   Újoinr  )rU   rZ   r[   r5   r5   r6   Ú_get_publish_topic‚  s   
zChannel._get_publish_topicc                 C   s   | j | \}}|  ||¡S rK   )r  r#  )rU   r  rZ   r[   r5   r5   r6   Ú_get_subscribe_topic‡  s   zChannel._get_subscribe_topicc                    sN   ‡ fdd„ˆ j D ƒ}|sd S ˆ j}|jjd u r|j ¡  |jˆ _| |¡ d S )Nc                    s   g | ]}ˆ   |¡‘qS r5   )r$  ©Ú.0r  r“   r5   r6   Ú
<listcomp>Œ  s    ÿz&Channel._subscribe.<locals>.<listcomp>)rË   rÄ   r¯   r«   r²   rÅ   Ú
psubscribe)rU   ÚkeysÚcr5   r“   r6   rÆ   ‹  s   
ÿ
zChannel._subscribec                 C   s6   |   |¡}| j}|jr|jjr| |g¡ d S d S d S rK   )r$  rÄ   r¯   r«   Úunsubscribe)rU   r  Útopicr*  r5   r5   r6   r   –  s
   
ÿzChannel._unsubscribe_fromc                 C   sŠ   t |d ƒdkr|d dkrd|_d S t |d ƒdkr.|d |d |d |d f\}}}}n|d d |d |d f\}}}}||||dœS )	Nr   r+  é   FÚpmessager   r   )r¶   ÚpatternrM   Údata)r   Ú
subscribed)rU   rE   Úrr¶   r/  rM   r0  r5   r5   r6   Ú_handle_messageœ  s   & üzChannel._handle_messagec                 C   sz   | j }g }z
| |  |¡¡ W n	 ty   Y nw |jd ur9|jjdd�r9| |  |¡¡ |jd ur9|jjdd�s%t|ƒS )Nr   r>   )rÄ   re   Ú_receive_oner   r¯   Úcan_readÚany)rU   r*  rà   r5   r5   r6   rÿ   ¬  s   ÿÿzChannel._receivec              	   C   sþ   d }z|  ¡ }W n | jy   d | _‚ w |d ury|  ||¡}t|d ƒ d¡r{t|d ƒ}|d r}|d dkr@| d¡\}}}z
tt|d ƒƒ}W n tt	fyd   t
d|t|ƒd d	… d
d� tƒ ‚w | dd
¡d }| j || j| ¡ dS d S d S d S )Nr¶   rf   rM   r0  r   r!  Ú.z&Cannot process event on channel %r: %si   r   r  T)Úparse_responser#   rÅ   r3  r   ÚendswithÚ	partitionr   r­   r£   ÚwarnÚreprr   Úsplitr¯   Ú_deliverrý   )rU   r*  Úresponser  rM   r�   rf   rZ   r5   r5   r6   r4  ·  s<   þÿý
ÿðzChannel._receive_oner   c                    s^   ˆj  tˆjƒ¡‰ ˆ sd S ‡ ‡fdd„ˆjD ƒ|pdg }ˆjjˆ_ˆjjjdg|¢R Ž  d S )Nc                    s"   g | ]}ˆ D ]}ˆ  ||¡‘qqS r5   )Ú
_q_for_pri)r&  Úprir  ©ÚqueuesrU   r5   r6   r'  Ô  s
    ÿÿz(Channel._brpop_start.<locals>.<listcomp>r   r¾   )	rø   ÚconsumeÚlenrÈ   rö   rE   r¯   r¿   Úsend_command)rU   r?   r)  r5   rB  r6   rÀ   Ð  s   ÿ
zChannel._brpop_startc                 K   sž   zJz| j j| j jdfi |¤Ž}W n | jy   | j j ¡  ‚ w |rH|\}}t|ƒ | jd¡d }| j 	|¡ | j 
tt|ƒƒ|¡ W d | _dS tƒ ‚d | _w )Nr¾   r   r   T)rE   r8  r¯   r#   r  r   Úrsplitrê   rø   Úrotater>  r   r¿   r   )rU   ÚoptionsÚ
dest__itemÚdestÚitemr5   r5   r6   rþ   Ù  s(   ÿ
þü
þzChannel._brpop_readc                 K   s,   |dkr| j  ¡  d S | j | jj|¡ d S )NrÃ   )rÄ   r8  rE   r¯   )rU   r¶   rI  r5   r5   r6   rÙ   ï  s   zChannel._poll_errorc                 C   sd   |   ¡ �$}| jD ]}| |  ||¡¡}|r$tt|ƒƒ  W  d   ƒ S qtƒ ‚1 s+w   Y  d S rK   )rn   rö   Úrpopr@  r   r   r   )rU   r  rE   rA  rL  r5   r5   r6   Ú_getõ  s   

üûzChannel._getc              	   C   sœ   |   ¡ �@}| ¡ �+}| jD ]}| |  ||¡¡}q| ¡ }tdd„ |D ƒƒW  d   ƒ W  d   ƒ S 1 s7w   Y  W d   ƒ d S 1 sGw   Y  d S )Nc                 s   s    � | ]}t |tjƒr|V  qd S rK   )r   ÚnumbersÚIntegral)r&  Úsizer5   r5   r6   Ú	<genexpr>  s   € 
ÿz Channel._size.<locals>.<genexpr>)rn   r{   rö   Úllenr@  rd   Úsum)rU   r  rE   rl   rA  Úsizesr5   r5   r6   Ú_sizeý  s   


üÿ"ÿzChannel._sizec                 C   s"   |   |¡}|rd || j|¡S |S )Nz{}{}{})ÚpriorityÚformatrê   )rU   r  rA  r5   r5   r6   r@    s   
zChannel._q_for_pric                 C   s   | j }|t||ƒd  S )Nr   )rö   r   )rU   ÚnÚstepsr5   r5   r6   rW    s   zChannel.priorityc                 K   sT   | j |dd�}|  ¡ �}| |  ||¡t|ƒ¡ W d  ƒ dS 1 s#w   Y  dS )zDeliver message.F)ÚreverseN)Ú_get_message_priorityrn   r  r@  r   )rU   r  rf   rW   rA  rE   r5   r5   r6   Ú_put  s   
"ÿzChannel._putc                 K   sF   |   ¡ �}| |  ||¡t|ƒ¡ W d  ƒ dS 1 sw   Y  dS )zDeliver fanout message.N)rn   Úpublishr#  r   )rU   rZ   rf   r[   rW   rE   r5   r5   r6   Ú_put_fanout  s   

þ"ÿzChannel._put_fanoutc                 K   s   |r
| j  |¡ d S d S rK   )rü   r§   )rU   r  Úauto_deleterW   r5   r5   r6   Ú
_new_queue  s   ÿzChannel._new_queuec              	   C   s‚   |   |¡jdkr|| dd¡f| j|< |  ¡ � }| | j|f | j |p%d|p(d|p+dg¡¡ W d   ƒ d S 1 s:w   Y  d S )NÚfanoutú#Ú*r÷   )	Útypeofr¶   Úreplacer  rn   ÚsaddÚkeyprefix_queuerê   r"  )rU   rZ   r[   r/  r  rE   r5   r5   r6   Ú_queue_bind#  s   
ÿ
þÿ"ÿzChannel._queue_bindc           
   	   O   sÐ   | j  |¡ | j| d¡d��O}| | j|f | j |pd|p d|p#dg¡¡ | ¡ �}| j	D ]}	| 
|  ||	¡¡}q/| ¡  W d   ƒ n1 sIw   Y  W d   ƒ d S W d   ƒ d S 1 saw   Y  d S )NrE   rm   r÷   )rü   rª   rn   rÓ   Úsremrh  rê   r"  r{   rö   Údeleter@  rd   )
rU   r  rZ   r[   r/  rV   rW   rE   rl   rA  r5   r5   r6   Ú_delete/  s    þÿ


ýû"ûzChannel._deletec              	   K   sŽ   |   ¡ �9}| ¡ �$}| jD ]}| |  ||¡¡}qt| ¡ ƒW  d   ƒ W  d   ƒ S 1 s0w   Y  W d   ƒ d S 1 s@w   Y  d S rK   )rn   r{   rö   Úexistsr@  r6  rd   )rU   r  rW   rE   rl   rA  r5   r5   r6   Ú
_has_queue;  s   



ýÿ"ÿzChannel._has_queuec                    sf   ˆ j | }ˆ  ¡ � }| |¡}|stt ||¡ƒ‚‡ fdd„|D ƒW  d   ƒ S 1 s,w   Y  d S )Nc                    s    g | ]}t t|ƒ ˆ j¡ƒ‘qS r5   )Útupler   r=  rê   )r&  Úvalr“   r5   r6   r'  H  s     z%Channel.get_table.<locals>.<listcomp>)rh  rn   Úsmembersr	   ÚNO_ROUTE_ERRORrX  )rU   rZ   ÚkeyrE   r   r5   r“   r6   Ú	get_tableB  s   


$üzChannel.get_tablec              	   C   s¦   |   ¡ �E}| ¡ �0}| jD ]}|  ||¡}| |¡ |¡}q| ¡ }t|d d d… ƒW  d   ƒ W  d   ƒ S 1 s<w   Y  W d   ƒ d S 1 sLw   Y  d S )Nr-  )rn   r{   rö   r@  rS  rk  rd   rT  )rU   r  rE   rl   rA  ÚpriqrU  r5   r5   r6   Ú_purgeJ  s   


ûÿ"ÿzChannel._purgec                    st   d| _ | js1| jj | ¡ | j d¡}|d ur)| jD ]}|| jv r(| j	||d� q|  
¡  |  ¡  tt| ƒ ¡  d S )NTrE   rm   )Ú_closingÚclosedr¯   r  rª   rÒ   rÓ   r  rü   Úqueue_deleter  Ú_close_clientsrQ   rå   r¥   )rU   rE   r  rX   r5   r6   r¥   S  s   

€zChannel.closec                 C   sL   dD ]!}z| j | }|jd }|_| ¡  W q tt| jfy#   Y qw d S )N)rE   rÄ   )rÒ   r¯   r  r¢   r¬   r4   )rU   ÚattrrE   r¯   r5   r5   r6   rz  c  s   
ÿûzChannel._close_clientsc                 C   sd   t |tjƒs0|r|dkrt}n| d¡r|dd … }zt|ƒ}W |S  ty/   td |¡ƒ‚w |S )Nr!  r   z0Database is int between 0 and limit - 1, not {0})r   rO  rP  Ú
DEFAULT_DBÚ
startswithÚintr£   rX  )rU   Úvhostr5   r5   r6   Ú_prepare_virtual_hostm  s    

ûÿÿÿzChannel._prepare_virtual_hostc                 K   s   |S rK   r5   )rU   rð   rñ   Úparamsr5   r5   r6   Ú_filter_tcp_connparams|  s   zChannel._filter_tcp_connparamsc                    s”  | j j}|jpd|jp| j j|j|j| j| j| j	| j
| j| j| jdœ}| j}t|dƒr6t|jdƒs6| d¡ |jrPz| |j¡ tj|d< W n	 tyO   Y nw |d }d|v r t|ƒ\}}}}}}	}
|dkr�| jdi |¤Ž}|jtjd	|	 d
œfi |
¤Ž | dd ¡ | dd ¡ | dd ¡ ||d< | dd ¡ | dd ¡ |  | dd ¡¡|d< | ‰| d¡p´| j}|rÄG ‡ ‡fdd„d|ƒ‰ ˆ }||d< |S )Nz	127.0.0.1)ÚhostÚportÚvirtual_hostÚpasswordró   rî   rï   rð   rñ   rô   rõ   rR   rô   Úconnection_classrƒ  z://r,   r!  )r‡  Úpathrï   rð   rñ   r†  r„  r…  Údbc                       s    e Zd Z‡‡ ‡fdd„Z‡  ZS )z'Channel._connparams.<locals>.Connectionc                    s   t ˆ | ƒ ¡  ˆ | ¡ d S rK   )rQ   r  r°   r“   )Ú
ConnectionrY   rM   r5   r6   r  ½  s   z2Channel._connparams.<locals>.Connection.disconnect)r:   r;   r<   r  r™   r5   ©rŠ  rM   rX   r6   rŠ  ¼  s    rŠ  r5   )r¯   rE   Úhostnamer„  Údefault_portr…  r†  ró   rî   rï   rð   rñ   rô   rõ   r‡  r)   r   rR   rÝ   ÚsslÚupdater(   ÚSSLConnectionr­   r   r‚  ÚUnixDomainSocketConnectionr€  rÓ   )rU   ÚasynchronousÚconninfoÚ
connparamsÚ
conn_classrƒ  Úschemer�   r†  rˆ  ÚqueryÚconnection_clsr5   r‹  r6   Ú_connparams€  sn   õÿ
þ
ÿþþ
ÿ
þzChannel._connparamsc                 C   s    |r	| j | jd�S | j | jd�S )N)rº   )rú   r
  r	  ©rU   r’  r5   r5   r6   Ú_create_clientÆ  rz   zChannel._create_clientc                 C   s0   | j |d�}| jj|d d�| _tjdi |¤ŽS )N©r’  r‰  )r‰  r5   )r™  r  rX  r(   ÚConnectionPool)rU   r’  r�  r5   r5   r6   Ú	_get_poolË  s   zChannel._get_poolc                 C   s   t jdk rtd t ¡ƒ‚t jS )N)r   r-  r   zSRedis transport requires redis-py versions 3.2.0 or later. You have {0.__version__})r(   r]   r
   rX  ÚStrictRedisr“   r5   r5   r6   rù   Ð  s   
þzChannel._get_clientc                 c   s   � |r|V  d S |   ¡ V  d S rK   ©r›  ©rU   rE   r5   r5   r6   rn   ×  s   €
zChannel.conn_or_acquirec                 C   s   | j d u r
|  ¡ | _ | j S rK   )r  rž  r“   r5   r5   r6   r	  Þ  s   

zChannel.poolc                 C   s   | j d u r| jdd�| _ | j S )NTrœ  )r  rž  r“   r5   r5   r6   r
  ä  s   
zChannel.async_poolc                 C   s   | j dd�S )z+Client used to publish messages, BRPOP etc.Trœ  r   r“   r5   r5   r6   rE   ê  s   zChannel.clientc                 C   s   | j dd�}| ¡ S )z1Pub/Sub connection used to consume fanout queues.Trœ  )r›  Úpubsubr¡  r5   r5   r6   rÄ   ï  s   zChannel.subclientc                 C   s   | j  | j¡ d S rK   )rø   r�  rÈ   r“   r5   r5   r6   r  õ  s   zChannel._update_queue_cyclec                 C   s   ddl m} |jS )Nr   r%   )r(   r&   r4   )rU   r&   r5   r5   r6   rû   ø  s   zChannel._get_response_errorc                    s   ‡ fdd„ˆ j D ƒS )z<Set of queues being consumed from (excluding fanout queues).c                    s   h | ]	}|ˆ j vr|’qS r5   )rË   r%  r“   r5   r6   Ú	<setcomp>ÿ  s    
ÿz(Channel.active_queues.<locals>.<setcomp>)Ú_active_queuesr“   r5   r“   r6   rÈ   ü  s   zChannel.active_queuesr—   r•   )r   r–   rK   )_r:   r;   r<   r=   rP   Ú_clientÚ
_subclientrw  Úsupports_fanoutrh  r  rê   r¿   rÅ   r  rë   rb   r`   r„   r…   rÎ   rƒ   ÚPRIORITY_STEPSrö   rî   rï   rð   rñ   rõ   ró   ÚDEFAULT_HEALTH_CHECK_INTERVALrô   rì   rí   rò   r  r  r   rå   Úfrom_transport_optionsr(   rŠ  r‡  rR   rL   r  r°   rŽ   r  r  r  r  r  r#  r$  rÆ   r   r3  rÿ   r4  rÀ   rþ   rÙ   rN  rV  r@  rW  r]  r_  ra  ri  rl  rn  rt  rv  r¥   rz  r€  r‚  r™  r›  rž  rù   r   rn   rä   r	  r
  r   rE   rÄ   r  rû   rÈ   r™   r5   r5   rX   r6   rå   Ž  s¾    ÿÿ%	
ÿ
		
	

ÿ

F




rå   c                       sn   e Zd ZdZeZdZeZdZdZ	e
jjjdeg d¢ƒd�Z‡ fdd„Zd	d
„ Zdd„ Zdd„ Zdd„ Z‡  ZS )r+   zRedis Transport.Nr(   T)Údirectr,  rb  )r’  Úexchange_typec                    sB   t d u rtdƒ‚tt| ƒj|i |¤Ž |  ¡ \| _| _tƒ | _	d S )Nz)Missing redis library (pip install redis))
r(   ÚImportErrorrQ   r+   rR   Ú_get_errorsr#   r$   rš   r  rT   rX   r5   r6   rR     s
   zTransport.__init__c                 C   s   t jS rK   )r(   Ú__version__r“   r5   r5   r6   Údriver_version  s   zTransport.driver_versionc                    s„   | j ‰ˆ ˆj¡ ˆj‰ˆj‰ | j‰‡fdd„}|ˆ_‡ ‡‡‡fdd„}ˆj |¡ ˆ 	dˆj
¡ |jj dt¡}ˆ 	|ˆj¡ d S )Nc                    s   | j rˆ  | j ¡ d S d S rK   )r«   r  )r¯   )Úloopr5   r6   Ú_on_disconnect&  s   ÿz:Transport.register_with_event_loop.<locals>._on_disconnectc                      s    ˆƒ  ‡ ‡fdd„ˆj D ƒ d S )Nc                    s   g | ]}ˆ |ˆ|ƒ‘qS r5   r5   )r&  r¤   )Ú
add_readerrØ   r5   r6   r'  -  s    zMTransport.register_with_event_loop.<locals>.on_poll_start.<locals>.<listcomp>)râ   r5   )r³  r  Úcycle_poll_startrØ   r5   r6   rÌ   +  s   z9Transport.register_with_event_loop.<locals>.on_poll_startr~   rô   )r  rÏ   rŸ   rÌ   r³  rØ   r°   Úon_tickr§   Úcall_repeatedlyrÐ   rE   Útransport_optionsrÓ   r©  rÕ   )rU   r¯   r±  r²  rÌ   rô   r5   )r³  r  r´  r±  rØ   r6   Úregister_with_event_loop  s$   þþz"Transport.register_with_event_loopc                 C   s   | j  |¡ dS )z1Handle AIO event for one of our file descriptors.N)r  rØ   )rU   r³   r5   r5   r6   rØ   9  s   zTransport.on_readablec                 C   s   t ƒ S )z3Utility to import redis-py's exceptions at runtime.)r7   r“   r5   r5   r6   r®  =  rã   zTransport._get_errors)r:   r;   r<   r=   rå   Úpolling_intervalÚDEFAULT_PORTr�  Údriver_typeÚdriver_namer   r+   Ú
implementsÚextendÚ	frozensetrR   r°  r¸  rØ   r®  r™   r5   r5   rX   r6   r+     s     
þ
r+   c                   @   s<   e Zd ZdZejd ZerejndZd	dd„Z	d	dd„Z
dS )
ÚSentinelChannelaï  Channel with explicit Redis Sentinel knowledge.

    Broker url is supposed to look like:

    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 :method:`Celery.connection` which is in charge of creating the
    `Connection` object).

    You must provide at least one option in Transport options:
     * `master_name` - name of the redis group to poll
    )Úmaster_nameÚmin_other_sentinelsÚsentinel_kwargsNFc                 C   sÀ   |   |¡}| ¡ }| dd ¡ | dd ¡ g }| jjjD ]}t|ƒ}|jdkr0| |j	|j
f¡ q|s>| |d |d f¡ tj|ft| ddƒt| dd ƒdœ|¤Ž}t| dd ƒ}| || j¡jS )	Nrƒ  r„  r   rÂ  r   rÃ  )rÂ  rÃ  rÁ  )r™  ÚcopyrÝ   r¯   rE   Úaltr   r–  re   rŒ  r„  r   ÚSentinelr¹   Ú
master_forrú   rº   )rU   r’  r”  Úadditional_paramsÚ	sentinelsÚurlÚsentinel_instrÁ  r5   r5   r6   Ú_sentinel_managed_poolZ  s6   

€ÿ

ýüþýz&SentinelChannel._sentinel_managed_poolc                 C   s
   |   |¡S rK   )rÌ  rš  r5   r5   r6   rž  y  s   
zSentinelChannel._get_poolr•   )r:   r;   r<   r=   rå   rª  r   ÚSentinelManagedConnectionr‡  rÌ  rž  r5   r5   r5   r6   rÀ  B  s    

rÀ  c                   @   s   e Zd ZdZdZeZdS )ÚSentinelTransportzRedis Sentinel Transport.ig  N)r:   r;   r<   r=   r�  rÀ  rå   r5   r5   r5   r6   rÎ  }  s    rÎ  )Gr=   Ú
__future__r   r   rO  r,   r   Úcollectionsr   Ú
contextlibr   r   Úviner   Úkombu.exceptionsr	   r
   Ú
kombu.fiver   r   r   Ú	kombu.logr   Úkombu.utils.compatr   Úkombu.utils.eventior   r   r   Úkombu.utils.encodingr   Úkombu.utils.jsonr   r   Úkombu.utils.objectsr   Úkombu.utils.schedulingr   Úkombu.utils.urlr   r   Úkombu.utils.functionalr   r÷   r   r(   r­  r   ÚloggerÚcriticalr;  r  rº  r|  r©  r¨  r"   rr  r7   r8   r  r9   rI   rN   rP   Úobjectrš   rå   r+   rÀ  rÎ  r5   r5   r5   r6   Ú<module>   sr    ÿÿ

h      y?;