
    D6i                       d Z ddlmZ ddl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 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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l0Z0 e ed             ed      k\  Z1	 ddl0m3Z3m4Z4  ed      Z5e5jl                  e5jn                  cZ8Z7dZ9dZ:dZ;g dZ< ed d!      Z=d" Z>d# Z? G d$ d%e@      ZAe
d&        ZBd' ZC G d( d)      ZD G d* d+eDe0j                        ZF G d, d-eDe0j                  j                        ZI G d. d/e0j                  j                        ZK G d0 d1e/j                        ZL G d2 d3      ZM G d4 d5e/j                        ZN G d6 d7e/j                        ZOe4r  G d8 d9e4j                  e0j                        ZR G d: d;eN      ZS G d< d=eO      ZTy# e2$ r dZ0dZ1Y Fw xY w# e2$ r dZ4dZ3Y Mw xY w)>a  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``
* ``client_name``: (str) The name to use when connecting to Redis server.
    )annotationsN)bisect)
namedtuple)contextmanager)version)Empty)time)Version)promise)InconsistencyErrorVersionMismatch)
get_logger)symbol_by_name)register_after_fork)bytes_to_str)ERRREADpoll)accepts_argument)dumpsloads)cached_property)cycle_by_name)
_parse_url   )virtualredisz5.3.0)CredentialProvidersentinelzkombu.transport.redisi     )r         	   error_classes_t)connection_errorschannel_errorsc                    ddl m}  t        | d      r| j                  }n| j                  }t        t        j                  j                  t        t        j                  t        t        | j                  | j                  | j                   | j"                  fz   t        j                  j$                  || j&                  | j(                  fz         S )z$Return tuple of redis error classes.r   
exceptionsInvalidData)r   r)   hasattrr*   	DataErrorr$   r   	Transportr%   r   socketerrorIOErrorOSErrorConnectionErrorBusyLoadingErrorAuthenticationErrorTimeoutErrorr&   InvalidResponseResponseError)r)   r,   s     X/home/azureuser/techstart-app/venv/lib/python3.12/site-packages/kombu/transport/redis.pyget_redis_error_classesr9      s      z=)**	((					,	,LL&&''**##0% 
% 
			)	)&&$$-& 
&     c                 &    ddl m}  | j                  S )z1Return the redis ConnectionError exception class.r   r(   )r   r)   r2   r(   s    r8   get_redis_ConnectionErrorr<      s     %%%r:   c                      e Zd ZdZy)	MutexHeldz)Raised when another party holds the lock.N__name__
__module____qualname____doc__ r:   r8   r>   r>      s    3r:   r>   c              #  f  K   | j                  ||      }d}	 |j                  d      }|rd n
t               	 |r	 |j                          yy# t        j
                  j                  $ r Y yw xY w# |r5	 |j                          w # t        j
                  j                  $ r Y w w xY ww xY ww)zTAcquire redis lock in non blocking way.

    Raise MutexHeld if not successful.
    timeoutF)blockingN)lockacquirer>   releaser   r)   LockNotOwnedError)clientnameexpirerI   lock_acquireds        r8   MutexrQ      s      ;;tV;,DMe4+    ##55  ##55  sc   B1#A5 B1 A B1A2/B11A22B15B.9B
	B.
B*'B.)B**B..B1c                $    | j                          y N)_after_fork)channels    r8   _after_fork_cleanup_channelrV      s    r:   c                  h     e Zd ZdZg dZddddddddddddd	Zd
 Z fdZ fdZddZ	 x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.
    )HDELHGETHLENHSETLLENLPUSHPUBLISHRPUSHRPOPSADDSREMSETSMEMBERSZADDZREMZREVRANGEBYSCOREr   N)
args_startargs_end   r!   )DELBRPOPEVALSHAWATCHc                   t        |      }|j                  d      }|| j                  v r| j                  t	        |d         z   |d<   nv|| j
                  v rh| j
                  |   d   }| j
                  |   d   }|dkD  r|d | ng }g }|||d  }|||| D cg c]  }| j                  t	        |      z    c}z   |z   }|g|S c c}w )Nr   ri   rj   )listpopPREFIXED_SIMPLE_COMMANDSglobal_keyprefixstrPREFIXED_COMPLEX_COMMANDS)selfargscommandri   rj   pre_args	post_argsargs           r8   _prefix_argsz!GlobalKeyPrefixMixin._prefix_args   s    Dz((1+d333++c$q'l:DG66677@NJ55g>zJH,6NtKZ(HI# O	
84 %%C0  D
 $s   Cc                z    t        |   ||fi |}|dk(  r#|r!|\  }}|t        | j                        d }||fS |S )zParse a response from the Redis server.

        Method wraps ``redis.parse_response()`` to remove prefixes of keys
        returned by redis command.
        rn   Nsuperparse_responselenru   )rx   
connectioncommand_nameoptionsretkeyvalue	__class__s          r8   r   z#GlobalKeyPrefixMixin.parse_response   sU     g$ZII7"sJCc$//012C:
r:   c                @    t        |   | j                  |      i |S rS   r   execute_commandr~   rx   ry   kwargsr   s      r8   r   z$GlobalKeyPrefixMixin.execute_command  #    w&(9(9$(?J6JJr:   c                ^    t        | j                  | j                  ||| j                        S )Nru   )PrefixedRedisPipelineconnection_poolresponse_callbacksru   )rx   transaction
shard_hints      r8   pipelinezGlobalKeyPrefixMixin.pipeline  s0    $  ##!22
 	
r:   )TN)r@   rA   rB   rC   rt   rw   r~   r   r   r   __classcell__r   s   @r8   rX   rX      sO     (  T2 !r2"#3 !t4	! .K
r:   rX   c                      e Zd ZdZd Zd Zy)PrefixedStrictRedisz@Returns a ``StrictRedis`` client that prefixes the keys it uses.c                x    |j                  dd      | _        t        j                  j                  | g|i | y Nru    )rs   ru   r   Redis__init__rx   ry   r   s      r8   r   zPrefixedStrictRedis.__init__  s2     &

+=r BT3D3F3r:   c                F    t        | j                  fd| j                  i|S )Nru   )PrefixedRedisPubSubr   ru   )rx   r   s     r8   pubsubzPrefixedStrictRedis.pubsub  s0    "  
!22
 
 	
r:   N)r@   rA   rB   rC   r   r   rD   r:   r8   r   r     s    J4
r:   r   c                      e Zd ZdZd Zy)r   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                    |j                  dd      | _        t        j                  j                  j
                  | g|i | y r   )rs   ru   r   rM   Pipeliner   r   s      r8   r   zPrefixedRedisPipeline.__init__)  s8     &

+=r B&&t=d=f=r:   N)r@   rA   rB   rC   r   rD   r:   r8   r   r   !  s    >r:   r   c                  @     e Zd ZdZdZ fdZd Z fdZ fdZ xZ	S )r   zCRedis pubsub client that takes global_keyprefix into consideration.)	SUBSCRIBEUNSUBSCRIBE
PSUBSCRIBEPUNSUBSCRIBEc                R    |j                  dd      | _        t        |   |i | y r   )rs   ru   r   r   r   s      r8   r   zPrefixedRedisPubSub.__init__8  s)     &

+=r B$)&)r:   c                    t        |      }|j                  d      }|| j                  v r%|D cg c]  }| j                  t	        |      z    }}|g|S c c}w Nr   )rr   rs   PUBSUB_COMMANDSru   rv   )rx   ry   rz   r}   s       r8   r~   z PrefixedRedisPubSub._prefix_args<  si    Dz((1+d***   %%C0D 
 $s   Ac                    t        |   |i |}||S |^}}}|g|D cg c]  }|t        | j                        d  c}|S c c}w )zParse a response from the Redis server.

        Method wraps ``PubSub.parse_response()`` to remove prefixes of keys
        returned by redis command.
        Nr   )	rx   ry   r   r   message_typechannelsmessagerU   r   s	           r8   r   z"PrefixedRedisPubSub.parse_responseH  sq     g$d5f5;J ,/(x
BJKwgc$//012K
 
 	
Ks   Ac                @    t        |   | j                  |      i |S rS   r   r   s      r8   r   z#PrefixedRedisPubSub.execute_command^  r   r:   )
r@   rA   rB   rC   r   r   r~   r   r   r   r   s   @r8   r   r   .  s(    MO*
 
,K Kr:   r   c                       e Zd ZdZdZ fdZ fdZddZ fdZd fd	Z	e
dd       Zdd	Zdd
ZddZed        Zed        Zed        Zed        Zed        Z xZS )QoSzRedis Ack Emulation.Tc                2    t        |   |i | d| _        y r   )r   r   _vrestore_countr   s      r8   r   zQoS.__init__g  s    $)&) r:   c           
        |j                   }|d   |d   }}t        j                  d   dk\  r|t               ig}nt               |g}| j	                         5 } |j
                  | j                  g| j                  | j                  |t        |j                  ||g            j                          t        | 5  ||       d d d        y # 1 sw Y   y xY w)Nexchangerouting_keyr   r!   )delivery_infor   VERSIONr	   pipe_or_acquirezaddunacked_index_keyhsetunacked_keyr   _rawexecuter   append)	rx   r   delivery_tagdeliveryEXRK	zadd_argspiper   s	           r8   r   z
QoS.appendk  s    ((*%x'>B==q &/0I.I!!# 	2tDIId,,9y9d&&W\\2r235GN7L1	2 	2 	2s   A-CCc                    | j                   j                  |      5 }| j                  D ]  }| j                  ||        	 d d d        | j                  j	                          y # 1 sw Y   $xY w)NrM   )rU   conn_or_acquire
_deliveredrestore_by_tagclear)rx   rM   tags      r8   restore_unackedzQoS.restore_unacked|  sc    \\))&1 	8V 8##C#78	8 		8 	8s   %A%%A.c                b    | j                  |      j                          t        |   |       y rS   )_remove_from_indicesr   r   ack)rx   r   r   s     r8   r   zQoS.ack  s&    !!,/779L!r:   c                    |r| j                  |d       n| j                  |      j                          t        |   |       y NT)leftmost)r   r   r   r   r   )rx   r   requeuer   s      r8   rejectz
QoS.reject  s<    t<%%l3;;=L!r:   c              #     K   |r| y | j                   j                  |      5 }|j                          d d d        y # 1 sw Y   y xY wwrS   )rU   r   r   )rx   r   rM   s      r8   r   zQoS.pipe_or_acquire  sC     J--f5 (oo''( ( (s   #AA	AA
Ac                    | j                  |      5 }|j                  | j                  |      j                  | j                  |      cd d d        S # 1 sw Y   y xY wrS   )r   zremr   hdelr   )rx   r   r   s      r8   r   zQoS._remove_from_indices  sM    !!$' 	=499T33\BT--|<	= 	= 	=s   6AAc           
        | xj                   dz  c_         | j                   dz
  |z  ry | j                  j                         5 }t               | j                  z
  }	 t        || j                  | j                        5  |j                  | j                  |d|xr ||d      }|xs g D ]  \  }}| j                  ||        	 d d d        d d d        y # 1 sw Y   xY w# t        $ r Y  w xY w# 1 sw Y   y xY w)Nr   r   T)startnum
withscores)r   rU   r   r	   visibility_timeoutrQ   unacked_mutex_keyunacked_mutex_expirezrevrangebyscorer   r   r>   )	rx   r   r   intervalrM   ceilvisibler   scores	            r8   restore_visiblezQoS.restore_visible  s   !  1$0\\))+ 	v6D333D	64#9#9446 9$55..a!me 6 GG '.m 9
U++C899	 	9 9  	 	sC   C0!C!=ACC!C	C!!	C-*C0,C--C00C9c                      fd} j                   j                  |      5 }|j                  | j                         d d d        y # 1 sw Y   y xY w)Nc                    | j                  j                        }| j                          j                  |        |r8t	        t        |            \  }}}j                  j                  ||||        y y rS   )hgetr   multir   r   r   rU   _do_restore_message)r   pMr   r   r   rx   r   s        r8   restore_transactionz/QoS.restore_by_tag.<locals>.restore_transaction  sh    		$**C0AJJL%%c40!,q/2	2r00BD(K r:   )rU   r   r   r   )rx   r   rM   r   r   s   `` ` r8   r   zQoS.restore_by_tag  sO    	L \\))&1 	FV2D4D4DE	F 	F 	Fs   AAc                .    | j                   j                  S rS   )rU   r   rx   s    r8   r   zQoS.unacked_key  s    ||'''r:   c                .    | j                   j                  S rS   )rU   r   r   s    r8   r   zQoS.unacked_index_key      ||---r:   c                .    | j                   j                  S rS   )rU   r   r   s    r8   r   zQoS.unacked_mutex_key  r   r:   c                .    | j                   j                  S rS   )rU   r   r   s    r8   r   zQoS.unacked_mutex_expire  s    ||000r:   c                .    | j                   j                  S rS   )rU   r   r   s    r8   r   zQoS.visibility_timeout  s    ||...r:   rS   FNN)r   
   r   )NF)r@   rA   rB   rC   restore_at_shutdownr   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   s   @r8   r   r   b  s    !2" "" ( (=
"F ( ( . . . . 1 1 / /r:   r   c                      e Zd ZdZeez  ZdZdZd Z	d Z
d Zd Zd Zd	 Zd
 Zd Zd Zd Zd Zd Zd Zd Zd Zd ZddZed        Zy)MultiChannelPollerz%Async I/O poller for Redis transport.FNc                z    t               | _        i | _        i | _        t	               | _        t               | _        y rS   )set	_channels_fd_to_chan_chan_to_sockr   poller
after_readr   s    r8   r   zMultiChannelPoller.__init__  s-    f%r:   c                @   | j                   j                         D ]  }	 | j                  j                  |         | j                  j                          | j                  j                          | j                   j                          y # t        t
        f$ r Y w xY wrS   )	r  valuesr	  
unregisterKeyError
ValueErrorr  r   r  )rx   fds     r8   closezMultiChannelPoller.close  s    $$++- 	B&&r*	
 	   "	 j) s   BBBc                :    | j                   j                  |       y rS   )r  addrx   rU   s     r8   r  zMultiChannelPoller.add  s    7#r:   c                :    | j                   j                  |       y rS   )r  discardr  s     r8   r  zMultiChannelPoller.discard  s    w'r:   c                z    	 | j                   j                  |j                         y # t        t        f$ r Y y w xY wrS   )r	  r  _sockAttributeError	TypeErrorrx   r   s     r8   _on_connection_disconnectz,MultiChannelPoller._on_connection_disconnect  s6    	KK"":#3#34	* 		s   %( ::c                   |||f| j                   v r| j                  |||       |j                  j                  |j                  j	                          |j                  j                  }||f| j
                  |j                         <   || j                   |||f<   | j                  j                  || j                         y rS   )
r  _unregisterr   r  connectr  filenor	  register
eventflags)rx   rU   rM   typesocks        r8   	_registerzMultiChannelPoller._register  s    VT"d&8&88Wfd3""*%%'  &&+2D/'6:GVT23T4??3r:   c                Z    | j                   j                  | j                  |||f          y rS   )r	  r  r  )rx   rU   rM   r#  s       r8   r  zMultiChannelPoller._unregister  s&    t117FD2IJKr:   c                    t        |dd       Ft        r |j                  j                         |_        n |j                  j                  d      |_        |j                  j
                  d uxr |||f| j                  v S )Nr   _)getattr"_REDIS_GET_CONNECTION_WITHOUT_ARGSr   get_connectionr   r  r  )rx   rU   rM   cmds       r8   _client_registeredz%MultiChannelPoller._client_registered  sw    6<.61$*$:$:$I$I$K!$*$:$:$I$I#$N!!!''t3 =&#&$*<*<<	>r:   c                    ||j                   df}| j                  ||j                   d      sd|_         | j                  |  |j                  s|j	                          yy)zEnable BRPOP mode for channel.rn   FN)rM   r-  _in_pollr%  _brpop_start)rx   rU   idents      r8   _register_BRPOPz"MultiChannelPoller._register_BRPOP  sX    0&&wH$GDNNE"  "  r:   c                    | j                  ||j                  d      s$d|_        | j                  ||j                  d       |j                  s|j	                          yy)zEnable LISTEN mode for channel.LISTENFN)r-  	subclient
_in_listenr%  
_subscriber  s     r8   _register_LISTENz#MultiChannelPoller._register_LISTEN  sR    &&w0A0A8L!&GNN7G$5$5x@!!  "r:   c                    | j                   D ]W  }|j                  r+|j                  j                         r| j	                  |       |j
                  sG| j                  |       Y y rS   )r  active_queuesqoscan_consumer2  active_fanout_queuesr8  r  s     r8   on_poll_startz MultiChannelPoller.on_poll_start!  sS    ~~ 	/G$$;;**,((1++%%g.	/r:   c                    || _         | j                  D ])  }|j                  j                  |j                        c S  y N)r   )r	  r  r;  r   unacked_restore_limit)rx   r	  rU   s      r8   on_poll_initzMultiChannelPoller.on_poll_init)  sA    ~~ 	G;;..11 /  	r:   c                    | j                   D ]6  }|j                  s|j                  j                  |j                        c S  y r@  )r  r:  r;  r   rA  r  s     r8   maybe_restore_messagesz)MultiChannelPoller.maybe_restore_messages0  sC    ~~ 	G$${{2255 3  	r:   c                    | j                   D ]G  }|j                  j                  d      }|!t        t	        |dd             s8|j                          I y )Nr5  check_health)r  __dict__getcallabler)  rF  )rx   rU   rM   s      r8   maybe_check_subclient_healthz/MultiChannelPoller.maybe_check_subclient_health8  sN    ~~ 	&G%%))+6F! !FG##%	&r:   c                    | j                   |   \  }}|j                  j                         r |j                  |           y y rS   )r  r;  r<  handlers)rx   r   chanr#  s       r8   on_readablezMultiChannelPoller.on_readable@  s<    %%f-
d88!DMM$! "r:   c                    |t         z  r| j                  |      | fS |t        z  r$| j                  |   \  }}|j	                  |       y y rS   )r   rN  r   r  _poll_error)rx   r   eventrM  r#  s        r8   handle_eventzMultiChannelPoller.handle_eventE  sM    4<##F+T11S[))&1JD$T" r:   c                   d| _         	 | j                  D ]W  }|j                  r+|j                  j	                         r| j                  |       |j                  sG| j                  |       Y | j                  j                  |      }|rc|D ]^  \  }}| j                  ||      }|s d| _         | j                  r0	 | j                  j                         } |        | j                  r/y y  | j                          t               # t        $ r Y y w xY w# d| _         | j                  r@	 | j                  j                         } |        n# t        $ r Y w w xY w| j                  r?w w xY w)NTF)_in_protected_readr  r:  r;  r<  r2  r=  r8  r	  r   rR  r
  rs   r  rD  r   )	rx   callbackrG   rU   eventsr   rQ  r   funs	            r8   rH  zMultiChannelPoller.getL  sR   "&	>> 3(({{..0,,W5//))'23 [[%%g.F%+ MFE++FE:C ',D#////--/C E // '')'M   	 ',D#////--/C E    //s[   AD AD )D >D	 .D 		DDE--EE-	EE-EE-+E-c                    | j                   S rS   )r  r   s    r8   fdszMultiChannelPoller.fdsj  s    r:   rS   )r@   rA   rB   rC   r   r   r"  rT  r
  r   r  r  r  r  r%  r  r-  r2  r8  r>  rB  rD  rJ  rN  rR  rH  propertyrY  rD   r:   r8   r  r    s    /J  J
 #$(4L>#!/&"
#<    r:   r  c                      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$dZ%dZ&e'jP                  jR                  dz   Z)e*re*jV                  ndZ,e*re*jZ                  ndZ. fdZ/d Z0d Z1d Z2	 dBdZ3dB fd	Z4d Z5 fdZ6d Z7 fdZ8d Z9d Z:d Z;d Z<d Z=d  Z>d! Z?dCd"Z@d# ZAd$ ZBd% ZCd& ZDd' ZEd( ZFd) ZGd* ZHdBd+ZId, ZJd- ZKd. ZLd/ ZMd0 ZN fd1ZOd2 ZPd3 ZQ	 	 dDd4ZRd5 ZSdBd6ZTdBd7ZUdBd8ZVd9 ZWeXdCd:       ZYeZd;        Z[eZd<        Z\e]d=        Z^e]d>        Z_d? Z`d@ ZaeZdA        Zb xZcS )EChannelzRedis Channel.NFTz_kombu.binding.%sz/{db}.zunackedunacked_indexunacked_mutexi,  i  r   r   round_robin)sepack_emulationr   r   r   r   r   rA  fanout_prefixfanout_patternsru   socket_timeoutsocket_connect_timeoutsocket_keepalivesocket_keepalive_optionsqueue_order_strategymax_connectionshealth_check_intervalretry_on_timeoutpriority_stepsclient_namec                n   t        |   |i | | j                  st        j                  | _        d| _         t        | j                               | _        | j                         | _
        | j                         | _        t               | _        t               | _        i | _        | j"                  | j$                  d| _        | j(                  j*                  | _        | j,                  r,t/        | j,                  t0              r| j,                  | _        nd| _        	 | j4                  j7                          | j(                  j<                  j?                  |        d| _        | j(                  j@                  | _         tB        tC        | tD               y y # t8        $ r | j;                           w xY w)NF)rn   r4  r   T)#r   r   rb  r   r   _registeredr   ri  _queue_cycle_get_clientClient_get_response_errorr7   r  r=  auto_delete_queues_fanout_to_queue_brpop_read_receiverL  r   brpop_timeoutrc  
isinstancerv   keyprefix_fanoutrM   ping	Exception_disconnect_poolscycler  r%   r   rV   r   s      r8   r   zChannel.__init__  s`   $)&)!!{{DH DM$*C*CDF&&(!557$'E!"%% ""&"2"2dmmL!__::$,,c2(,(:(:% %'D!	KK
 	!!$' "&!B!B*&AB +  	""$	s    F F4c                $    | j                          y rS   )r~  r   s    r8   rT   zChannel._after_fork  s     r:   c                    | j                   }| j                  }d x| _        | _         ||j                          ||j                          y y rS   )_pool_async_pool
disconnect)rx   pool
async_pools      r8   r~  zChannel._disconnect_pools  sL    zz%%
(,,4:OO!!!# "r:   c                    | j                   |u rd | _         | j                  |u rd | _        | j                  r=| j                  j                  r&| j                  j                  j	                  |       y y y rS   )r/  r6  r   r  r  r  s     r8   r  z!Channel._on_connection_disconnect  sZ    ==J& DM??j("DO??t44OO!!;;JG  5?r:   c                N   	 	 d|d   d<   d|d   d   d<   | j                  ||      D ]O  }| j                  |d      } |r|j                  n|j                  | j                  ||      t        |             Q y # t         $ r Y pw xY w# t        $ r t        d|d	       Y y w xY w)
NTheadersredelivered
propertiesr   FreversezCould not restore message: %rexc_info)	r  _lookup_get_message_prioritylpushrpush
_q_for_prir   r}  crit)rx   payloadr   r   r   r   queuepris           r8   r   zChannel._do_restore_message  s    	J48	"=1HL%o6}E h< 00%0H7xTZZOOE3/w    	J0'DI	Js/   A; A$B
 ;	BB
 BB
 
B$#B$c                      j                   st         	  |      S |j                   fd} j	                         5 }|j                  | j                         d d d        y # 1 sw Y   y xY w)Nc                    | j                  j                        }| j                          | j                  j                         |r.t	        t        |            \  }}}j                  ||||        y y rS   )r   r   r   r   r   r   r   )r   Pr   r   r   r   rx   r   s        r8   r   z-Channel._restore.<locals>.restore_transaction2  sh    		$**C0AJJLIId&&,!,q/2	2r((BD(C r:   )rb  r   _restorer   r   r   r   )rx   r   r   r   rM   r   r   s   ` `  @r8   r  zChannel._restore-  sm    !!7#G,,""	D !!# 	Fv2D4D4DE	F 	F 	Fs   A))A2c                (    | j                  |d      S r   )r  )rx   r   s     r8   _restore_at_beginningzChannel._restore_at_beginning=  s    }}Wt}44r:   c                    || j                   v r<| j                   |   \  }}| j                  j                  |       || j                  |<   t	        |   |g|i |}| j                          |S rS   )_fanout_queuesr=  r  rv  r   basic_consume_update_queue_cycle)rx   r  ry   r   r   r(  r   r   s          r8   r  zChannel.basic_consume@  su    D'''--e4KHa%%))%0.3D!!(+g#E;D;F; 	  "
r:   c                    | j                   }|ra|j                  j                  r:|j                  j                  j	                  t        | j                  |f            S | j                  |      S y rS   )r   r  rT  r
  r  r   _basic_cancel)rx   consumer_tagr   s      r8   basic_cancelzChannel.basic_cancelT  sh    
 __
22!''2266D..@  %%l33 r:   c                v   	 | j                   |   }	 | j                  j                  |       | j	                  |       	 | j
                  |   \  }}| j                  j                  |       t        | %  |      }| j                          |S # t        $ r Y y w xY w# t        $ r Y iw xY w# t        $ r Y Jw xY wrS   )_tag_to_queuer  r=  remove_unsubscribe_fromr  rv  rs   r   r  r  )rx   r  r  r   r(  r   r   s         r8   r  zChannel._basic_cancela  s    	&&|4E	*%%,,U3 ""5)	--e4KHa!!%%h/ g"<0  "
  		  		  		s4   B B  -B, 	BB	B)(B),	B87B8c                    |r+| j                   rdj                  | j                  |d|g      S dj                  | j                  |g      S )Nr   /)rd  joinr{  )rx   r   r   s      r8   _get_publish_topiczChannel._get_publish_topicu  sE    4//77D118S+NOOww--x899r:   c                J    | j                   |   \  }}| j                  ||      S rS   )r  r  )rx   r  r   r   s       r8   _get_subscribe_topiczChannel._get_subscribe_topicz  s*     $ 3 3E :+&&x==r:   c                    | j                   D cg c]  }| j                  |       }}|sy | j                  }|j                  j                  |j                  j                          |j                  | _        |j                  |       y c c}w rS   )r=  r  r5  r   r  r  r6  
psubscribe)rx   r  keyscs       r8   r7  zChannel._subscribe~  s{    !668 ))%0 8 8NN<<%LL  ",,	T8s   Bc                    | j                  |      }| j                  }|j                  r*|j                  j                  r|j	                  |g       y y y rS   )r  r5  r   r  unsubscribe)rx   r  topicr  s       r8   r  zChannel._unsubscribe_from  sC    ))%0NN<<ALL..MM5'" /<r:   c                    t        |d         dk(  r|d   dk(  rd|_        y t        |d         dk(  r|d   |d   |d   |d   f\  }}}}n|d   d |d   |d   f\  }}}}||||dS )	Nr   r  rl   Fpmessager   r!   )r#  patternrU   data)r   
subscribed)rx   rM   rr#  r  rU   r  s          r8   _handle_messagezChannel._handle_message  s    !.1Q419 %F!++,Q41qtQqT+A(D'7D+,Q4qtQqT+A(D'7D	
 	
r:   c                v   | j                   }g }	 |j                  | j                  |             |j                  e|j                  j                  d      rI|j                  | j                  |             |j                  |j                  j                  d      rIt        |      S # t        $ r Y w xY w)Nr   rF   )r5  r   _receive_oner   r   can_readany)rx   r  r   s      r8   rx  zChannel._receive  s    NN	JJt((+, ll&1<<+@+@+@+KJJt((+, ll&1<<+@+@+@+K3x	  		s    B, ,	B87B8c                l   d }	 |j                         }t        |t        t
        f      r| j                  ||      }t        |d         j                  d      rt        |d         }|d   rt|d   dk(  r|j                  d      \  }}}	 t        t        |d               }|j!                  dd
      d   }| j"                  j%                  || j&                  |          yy y y # | j                  $ r	 d | _         w xY w# t        t        f$ r& t        d|t        |      d d	 d
       t               w xY w)Nr#  r   rU   r  r   r  .z&Cannot process event on channel %r: %si   r   r  T)r   r%   r6  rz  rr   tupler  r   endswith	partitionr   r  r  warningreprr   splitr   _deliverrv  )rx   r  responser  rU   r(  r   r   s           r8   r  zChannel._receive_one  sK   	'')H hu.**1h7GGFO,55i@&wy'9:6?qzS((/(9(9#(>1g&"'WV_(E"F
  '}}S!4Q7HOO,,!6!6x!@B # A / %% 	"DO	 &z2 & H 'gu)=K#g&s   C# 
C> #C;>5D3c           	        || j                   }| j                  j                  t        | j                              }|sy | j
                  D cg c]  }|D ]  }| j                  ||        c}}|xs dgz   }| j                  j                  | _	        dg|}| j                  r| j                  j                  |      } | j                  j                  j                  |  y c c}}w )Nr   rn   )ry  rq  consumer   r:  rm  r  rM   r   r/  ru   r~   send_command)rx   rG   queuesr  r  r  command_argss          r8   r0  zChannel._brpop_start  s    ?((G""**3t/A/A+BC7;7J7J %#% s+ %+ %(/1~6..'$'  ;;33LAL+++\:%s    C&c                   	 	  | j                   j                  | j                   j                  dfi |}|r|\  }}t        |      j                  | j                  d      d   }| j                  j                  |       | j                  j                  t        t        |            |       	 d | _        yt               # | j                  $ r& | j                   j                  j	                           w xY w# d | _        w xY w)Nrn   r   r   T)rM   r   r   r%   r  r   rsplitra  rq  rotater  r   r/  r   )rx   r   
dest__itemdestitems        r8   rw  zChannel._brpop_read  s    	!7T[[778N8N8?C:AC
 '
d#D)001=a@!!((.((|D/A)BDI !DM g ))  &&113	 !DMs#   2C  A8C8 6
C8  5C55C8 8	Dc                    |dk(  r| j                   j                          y | j                  j                  | j                  j                  |       y )Nr4  )r5  r   rM   r   )rx   r#  r   s      r8   rP  zChannel._poll_error  s8    8NN))+KK&&t{{'='=tDr:   c                    | j                         5 }| j                  D ]D  }|j                  | j                  ||            }|s't	        t        |            c cd d d        S  t               # 1 sw Y   y xY wrS   )r   rm  rpopr  r   r   r   )rx   r  rM   r  r  s        r8   _getzChannel._get  st    !!# 	v** 5{{4??5##>? d!344		 	5 'M	 	s   3A/A/$A//A8c           	     V   | j                         5 }|j                         5 }| j                  D ]#  }|j                  | j	                  ||            }% |j                         }t        d |D              cd d d        cd d d        S # 1 sw Y   nxY w	 d d d        y # 1 sw Y   y xY w)Nc              3  T   K   | ]   }t        |t        j                        r| " y wrS   )rz  numbersIntegral).0sizes     r8   	<genexpr>z Channel._size.<locals>.<genexpr>  s)      BD(w/?/?@   Bs   &()r   r   rm  llenr  r   sum)rx   r  rM   r   r  sizess         r8   _sizezChannel._size  s    !!# 	Bv" Bd.. BC99T__UC%@ADB BE B B	B B	B 	BB B B	B 	B 	Bs#   BAB	6	B	B	BB(c                P    | j                  |      }|r| | j                   | S |S rS   )priorityra  )rx   r  r  s      r8   r  zChannel._q_for_pri   s/    mmC WTXXJse,,r:   c                >    | j                   }|t        ||      dz
     S )Nr   )rm  r   )rx   nstepss      r8   r  zChannel.priority  s$    ##VE1%)**r:   c                    | j                  |d      }| j                         5 }|j                  | j                  ||      t	        |             ddd       y# 1 sw Y   yxY w)zDeliver message.Fr  N)r  r   r  r  r   )rx   r  r   r   r  rM   s         r8   _putzChannel._put
  s[    ((%(@!!# 	FvLL4eGnE	F 	F 	Fs   ,AA"c                    | j                         5 }|j                  | j                  ||      t        |             ddd       y# 1 sw Y   yxY w)zDeliver fanout message.N)r   publishr  r   )rx   r   r   r   r   rM   s         r8   _put_fanoutzChannel._put_fanout  sH    !!# 	vNN''+>g	 	 	s   ,AAc                @    |r| j                   j                  |       y y rS   )ru  r  )rx   r  auto_deleter   s       r8   
_new_queuezChannel._new_queue  s    ##''. r:   c           
     \   | j                  |      j                  dk(  r!||j                  dd      f| j                  |<   | j	                         5 }|j                  | j                  |fz  | j                  j                  |xs d|xs d|xs dg             d d d        y # 1 sw Y   y xY w)Nfanout#*r   )	typeofr#  replacer  r   saddkeyprefix_queuera  r  )rx   r   r   r  r  rM   s         r8   _queue_bindzChannel._queue_bind  s    ;;x %%1 +--c37*D& !!# 	6vKK,,{:{'8b'.}"',{'4 56	6 	6 	6s   A	B""B+c           
        | j                   j                  |       | j                  |j                  d            5 }|j	                  | j
                  |fz  | j                  j                  |xs d|xs d|xs dg             |j                         5 }| j                  D ]#  }	|j                  | j                  ||	            }% |j                          d d d        d d d        y # 1 sw Y   xY w# 1 sw Y   y xY w)NrM   r   r   )ru  r  r   rH  sremr  ra  r  r   rm  deleter  r   )
rx   r  r   r   r  ry   r   rM   r   r  s
             r8   _deletezChannel._delete)  s    ''.!!H)=!> 	&KK,,{:{'8b'.}"',{'4 56 " d.. DC;;tuc'BCDD	 	
 	 	s%   AC6AC*C6*C3	/C66C?c           	     D   | j                         5 }|j                         5 }| j                  D ]#  }|j                  | j	                  ||            }% t        |j                               cd d d        cd d d        S # 1 sw Y   nxY w	 d d d        y # 1 sw Y   y xY wrS   )r   r   rm  existsr  r  r   )rx   r  r   rM   r   r  s         r8   
_has_queuezChannel._has_queue5  s    !!# 	+v" +d.. DC;;tuc'BCDD4<<>*+ +	+ 	++ + +	+ 	+ 	+s#   BAB -	B B		BBc           	     (   | j                   |z  }| j                         5 }|j                  |      }|sg cd d d        S |D cg c]/  }t        t	        |      j                  | j                              1 c}cd d d        S c c}w # 1 sw Y   y xY wrS   )r  r   smembersr  r   r  ra  )rx   r   r   rM   r  vals         r8   	get_tablezChannel.get_table<  s    ""X-!!# 	Pv__S)F 	P 	P IOOE,s+11$((;<O	P 	P P	P 	Ps"   BB4B7BBBc                v   | j                         5 }|j                         5 }| j                  D ]4  }| j                  ||      }|j	                  |      j                  |      }6 |j                         }t        |d d d         cd d d        cd d d        S # 1 sw Y   nxY w	 d d d        y # 1 sw Y   y xY w)Nrl   )r   r   rm  r  r  r  r   r  )rx   r  rM   r   r  priqr  s          r8   _purgezChannel._purgeF  s    !!# 	'v" 'd.. 8C??5#6D99T?11$7D8 51:' '	' 	'' ' '	' 	' 	's#   B/A$B	B/B"	B//B8c                   d| _         | j                  r	 | j                          | j                  s| j
                  j                  j                  |        | j                  j                  d      }|3| j                  D ]$  }|| j                  v s| j                  ||       & | j                          | j                          t        | A          y # t        $ r Y w xY w)NTrM   r   )_closingr/  rw  r   closedr   r  r  rG  rH  r  ru  queue_deleter~  _close_clientsr   r  )rx   rM   r  r   s      r8   r  zChannel.closeO  s    ==  " {{OO!!))$/ ]]&&x0F!!00 @E 7 77))%)?@ ""$!  s   C 	C"!C"c                    dD ]6  }	 | j                   |   }|j                  d c}|_        |j                          8 y # t        t        | j
                  f$ r Y Vw xY w)N)rM   r5  )rG  r   r  r  r  r7   )rx   attrrM   r   s       r8   r  zChannel._close_clientsd  sf    ) 	Dt,060A0A4-
F-%%'		
 nd.@.@A s   3=AAc                    t        |t        j                        s2|r|dk(  rt        }n|j	                  d      r|dd  }	 t        |      }|S |S # t        $ r t        dj                  |            w xY w)Nr  r   z/Database is int between 0 and limit - 1, not {})rz  r  r  
DEFAULT_DB
startswithintr  format)rx   vhosts     r8   _prepare_virtual_hostzChannel._prepare_virtual_hostn  s    %!1!12ESL"!!#&ab	E
 u   ELL s    A $A3c                    |S rS   rD   )rx   rg  rh  paramss       r8   _filter_tcp_connparamszChannel._filter_tcp_connparams}  s    r:   c                    |rgt        |t              rt        |      } |       }t        |t              st	        d      ||d<   |j                  dd        |j                  dd        y y )NzRCredential provider is not an instance of a redis.CredentialProvider or a subclasscredential_providerusernamepassword)rz  rv   r   r   r  rs   )rx   r  
connparamscredential_provider_clss       r8   _process_credential_providerz$Channel._process_credential_provider  sp    -s3*89L*M'&=&?#13EF h  1DJ,-NN:t,NN:t, r:   c                   | j                   j                  }|j                  xs d|j                  xs | j                   j                  |j
                  |j                  |j                  |j                  | j                  | j                  | j                  | j                  | j                  | j                  | j                  | j                   d}| j#                  |j                  |       | j$                  }t'        |d      rW|g}t'        |d      r|t)        |j*                        z  }|D ]  }t-        |j.                  d      s n |j1                  d       |j2                  r+	 |j5                  |j2                         | j6                  |d<   |d   }d|v rt;        |      \  }}	}	}
}}}|d	k(  ro | j<                  di |} |j4                  t>        j@                  d
|z   dfi | |j1                  dd        |j1                  dd        |j1                  dd        |
|d<   ||d<   |j1                  dd       }| j#                  ||       |j1                  dd        |j1                  dd        | jC                  |j1                  dd             |d<   | |jE                  d      xs | j$                  }|r G fdd|      }|}||d<   |S # t8        $ r Y Kw xY w)Nz	127.0.0.1)hostportvirtual_hostr  r  r  rj  re  rf  rg  rh  rk  rl  rn  r   	__bases__rk  connection_classr$  z://r.   r  )r(  pathrf  rg  rh  r  r  r  r%  r&  dbc                  "     e Zd Z fdZ xZS )'Channel._connparams.<locals>.Connectionc                Z    t        |   |  j                  rj                  |        y y rS   )r   r  rp  r  )rx   ry   r   rU   s     r8   r  z2Channel._connparams.<locals>.Connection.disconnect  s.    G&- **99$? +r:   )r@   rA   rB   r  r   )r   rU   s   @r8   
Connectionr,    s    @ @r:   r.  rD   )#r   rM   hostnamer%  default_portr&  useridr  r  rj  re  rf  rg  rh  rk  rl  rn  r"  r(  r+   rr   r'  r   r   rs   sslupdateconnection_class_sslr  r   r  r   UnixDomainSocketConnectionr  rH  )rx   asynchronousconninfor   
conn_classclassesklassr$  schemer(  r  r  r)  queryr  connection_clsr.  rU   s                    @r8   _connparamszChannel._connparams  s   ??))%%4MMAT__%A%A$11  ))#+#?#?#33"11&*&A&A $ 5 5(,(E(E%)%?%? $ 5 5++

" 	))(*F*F
S**
 :z* "lGz;/4
 4 455  8#ENN4KL8 67<<!!(,,/151J1J
-. &!D=<Ft<L9FAq(HdE!8T88F:F
!
!!(-(H(H$J#( 2+02 7>14894@%-Jz"%-Jz" #()),A4"H--.A:NNN64(NN64(55NN>402
4 NN-. "!! 	
 @^ @ (N)7
%&Y  s   )*K 	KKc                v    |r| j                  | j                        S | j                  | j                        S )N)r   )rs  r  r  )rx   r6  s     r8   _create_clientzChannel._create_client  s/    ;;t;??{{499{55r:   c                    | j                  |      }| j                  j                  |d         | _        t        j                  di |S )Nr6  r*  r*  rD   )r>  r{  r  r   ConnectionPoolrx   r6  r  s      r8   	_get_poolzChannel._get_pool  sI    !!|!< $ 5 5 < <t < M##-f--r:   c                    t         j                  dk  rt        dj                  t                     | j                  r%t        j                  t        | j                        S t         j                  S )N)r!   rl   r   zSRedis transport requires redis-py versions 3.2.0 or later. You have {0.__version__}r   )	r   r   r   r  ru   	functoolspartialr   r   r   s    r8   rr  zChannel._get_client  s_    ==9$!++16%=: :   $$#!%!6!6 
 {{r:   c              #  >   K   |r| y | j                          y wrS   r@  rx   rM   s     r8   r   zChannel.conn_or_acquire  s     L%%''s   c                \    | j                   | j                         | _         | j                   S rS   )r  rF  r   s    r8   r  zChannel.pool	  s$    ::)DJzzr:   c                `    | j                   | j                  d      | _         | j                   S )NTrB  )r  rF  r   s    r8   r  zChannel.async_pool  s-    ##~~4~@Dr:   c                &    | j                  d      S )z+Client used to publish messages, BRPOP etc.TrB  rK  r   s    r8   rM   zChannel.client  s     """55r:   c                F    | j                  d      }|j                         S )z1Pub/Sub connection used to consume fanout queues.TrB  )r@  r   rL  s     r8   r5  zChannel.subclient  s#     $$$$7}}r:   c                N    | j                   j                  | j                         y rS   )rq  r3  r:  r   s    r8   r  zChannel._update_queue_cycle   s      !3!34r:   c                &    ddl m} |j                  S )Nr   r(   )r   r)   r7   )rx   r)   s     r8   rt  zChannel._get_response_error#  s    $'''r:   c                \    | j                   D ch c]  }|| j                  vr| c}S c c}w )z<Set of queues being consumed from (excluding fanout queues).)_active_queuesr=  )rx   r  s     r8   r:  zChannel.active_queues'  s7     $(#6#6 ;% 9 99  ; 	; ;s   )r   rS   r   )dr@   rA   rB   rC   r   _client
_subclientr  supports_fanoutr  r{  ra  r/  r6  r  rb  r   r   r   r   rA  r   PRIORITY_STEPSrm  re  rf  rg  rh  rl  rj  DEFAULT_HEALTH_CHECK_INTERVALrk  rn  rc  rd  ru   ri  r  r  r   r\  from_transport_optionsr   r.  r(  SSLConnectionr4  r   rT   r~  r  r   r  r  r  r  r  r  r  r7  r  r  rx  r  r0  rw  rP  r  r  r  r  r  r  r  r  r   r  r  r
  r  r  r  r  r"  r>  r@  rF  rr  r   r   rZ  r  r  r   rM   r5  r  rt  r:  r   r   s   @r8   r\  r\  o  sg   
CGJHO)O
CHJNMK'' #NN!#O9K M O
 4 )KE 	..		 2 ,1u''d275..T'CR!
$H ,1J"F 5(4(:
>	#
 	 2; !,EB+F/
6
+P'* 7;8<- Wr6
.
 ( (  
    
 6 6  
5( ; ;r:   r\  c                       e Zd ZdZeZdZdZeZdZ	dZ
ej                  j                  j                  d eg d            Zer
 e       \  ZZ fdZd	 Zd
 Zd Z xZS )r-   zRedis Transport.Nr   r   T)directr  r  )r6  exchange_typec                    t         t        d      t        |   |i | t	               | _        | j                  | j                  | _        y y )Nz)Missing redis library (pip install redis))r   ImportErrorr   r   r  r  polling_intervalry  r   s      r8   r   zTransport.__init__A  sQ    =IJJ$)&) ()
  ,!%!6!6D -r:   c                "    t         j                  S rS   )r   __version__r   s    r8   driver_versionzTransport.driver_versionL  s       r:   c                  	 | j                   j                  j                         j                  j                  | j
                  	fd}|_        	fdj                  j                         j                  dj                         |j                  j                  j                  dt              }j                  |j                         y )Nc                    | j                   rj                  | j                          j                  r	 j                  j                         y y # t        $ r Y y w xY wrS   )r  r  rY  on_tickr  )r   r  loopr>  s    r8   _on_disconnectz:Transport.register_with_event_loop.<locals>._on_disconnectV  sW    J,,- yyLL''6    s   A 	AAc                 b             j                   D  cg c]  }  | |        c}  y c c} w rS   )rY  )r  
add_readerr  cycle_poll_startrN  s    r8   r>  z9Transport.register_with_event_loop.<locals>.on_poll_startc  s'    7<yyAZK,AAs   ,r   rk  )r  rB  r	  r>  rk  rN  r  rg  r  call_repeatedlyrD  rM   transport_optionsrH  rY  rJ  )
rx   r   rh  ri  rk  rk  r  rl  r>  rN  s
     `  @@@@@r8   register_with_event_loopz"Transport.register_with_event_loopO  s    

4;;' ..__
&&
	 +9'	B 	'R!=!=> * 1 1 C C G G#)!
 	!..	
r:   c                :    | j                   j                  |       y)z1Handle AIO event for one of our file descriptors.N)r  rN  )rx   r   s     r8   rN  zTransport.on_readableq  s    

v&r:   )r@   rA   rB   rC   r\  ra  ry  DEFAULT_PORTr0  driver_typedriver_namer   r-   
implementsextend	frozensetr   r9   r%   r&   r   rd  ro  rN  r   r   s   @r8   r-   r-   .  sz    GMLKK""--44 => 5 J
 ,C,E)>	7! 
D'r:   r-   c                      e Zd ZdZy)SentinelManagedSSLConnectionzConnect 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.
        Nr?   rD   r:   r8   rx  rx  w  s    	 	r:   rx  c                  j    e Zd ZdZej
                  dz   Zerej                  ndZere	ndZ
ddZddZy)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_kwargsNc           	        | j                  |      }|j                         }|j                  dd        |j                  dd        g }| j                  j                  j
                  D ]^  }t        |      }|j                  dk(  s|j                  xs | j                  j                  }|j                  |j                  |f       ` |s|j                  |d   |d   f       t        j                  |ft        | dd      t        | dd       d|}t        | dd       }|t        d	      |j!                  |t"        j$                        j&                  S )
Nr$  r%  r   r|  r   r}  )r|  r}  r{  z1'master_name' transport option must be specified.)r>  copyrs   r   rM   altr   r;  r%  r0  r   r/  r   Sentinelr)  r  
master_forr   r   r   )	rx   r6  r   additional_params	sentinelsurlr%  sentinel_instr{  s	            r8   _sentinel_managed_poolz&SentinelChannel._sentinel_managed_pool  sQ   %%l3
&OO-fd+fd+	??))-- 	7CS/CzzZ'xx?4??#?#?  #,,!56		7 j0*V2DEF ))! '.CQ G#D*;TB!  	! dM48C  ''KK
 /	r:   c                    | j                  |      }| j                  j                  |d         | _        | j                  |      S )NrB  r*  rC  )r>  r{  r  r  rE  s      r8   rF  zSentinelChannel._get_pool  sF    !!|!< $ 5 5 < <t < M**<88r:   r   )r@   rA   rB   rC   r\  rZ  r   SentinelManagedConnectionr(  rx  r4  r  rF  rD   r:   r8   rz  rz    sH    8 %;; ? 
 >Fx994;C7#J9r:   rz  c                      e Zd ZdZdZeZy)SentinelTransportzRedis Sentinel Transport.ig  N)r@   rA   rB   rC   r0  rz  r\  rD   r:   r8   r  r    s    #LGr:   r  )UrC   
__future__r   rH  r  r.   r   collectionsr   
contextlibr   importlib.metadatar   r  r   r	   packaging.versionr
   viner   kombu.exceptionsr   r   	kombu.logr   kombu.utilsr   kombu.utils.compatr   kombu.utils.encodingr   kombu.utils.eventior   r   r   kombu.utils.functionalr   kombu.utils.jsonr   r   kombu.utils.objectsr   kombu.utils.schedulingr   kombu.utils.urlr   r   r   r   r*  r`  r   r   loggercriticalr  r  rq  r  rY  rX  r$   r9   r<   r}  r>   rQ   rV   rX   r   r   rM   r   r   PubSubr   r   r  r\  r-   r  r[  rx  rz  r  rD   r:   r8   <module>r     s  4l #     " % &   %  @   & 2 - / / 3 ) / 0 & .)01A)BggFV)V&
2 
+	,g
 " . 1 (4&4	 4  ,P
 P
f
. 

>0%,,2G2G 
>1K%,,-- 1Khh/'++ h/V_  _ D|
;goo |
;~E'!! E'P 
..
M9g M9`	 {+  .E)-&.  Hs$   G+ 2G; +	G87G8;	HH