o
    wvXj;6  ã                   @   sô   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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mZ ddlmZ ddlmZ ddlmZ dZdZG dd„ deƒZ G dd„ dej!ƒZ!G dd„ dej"ƒZ"dS )zxMongoDB transport.

:copyright: (c) 2010 - 2013 by Flavio Percoco Premoli.
:license: BSD, see LICENSE for more details.
é    )Úabsolute_importÚunicode_literalsN)Úerrors)ÚMongoClientÚ
uri_parser)Ú
CursorType)ÚVersionMismatch)ÚEmptyÚstring_t)Ú_detect_environment)Úbytes_to_str)ÚloadsÚdumps)Úcached_propertyé   )Úvirtual©Úto_rabbitmq_queue_argumentsz3Kombu requires MongoDB version 1.3+ (server is {0})zKKombu requires MongoDB version 2.2+ (server is {0}) for TTL indexes supportc                   @   sF   e Zd ZdZdd„ Zdd„ Zdd„ Zdd	d
„Zdd„ Zdd„ Z	e	Z
dS )ÚBroadcastCursorzCursor for broadcast queues.c                 C   s   || _ | jdd� d S )NF)Úrewind)Ú_cursorÚpurge)ÚselfÚcursor© r   úT/var/www/html/myproject/venv/lib/python3.10/site-packages/kombu/transport/mongodb.pyÚ__init__&   s   zBroadcastCursor.__init__c                 C   s   | j  ¡ | j S ©N)r   ÚcountÚ_offset©r   r   r   r   Úget_size+   ó   zBroadcastCursor.get_sizec                 C   s   | j  ¡  d S r   )r   Úcloser    r   r   r   r#   .   s   zBroadcastCursor.closeTc                 C   s.   |r| j  ¡  | j  ¡ | _| j  | j¡| _ d S r   )r   r   r   r   Úskip)r   r   r   r   r   r   1   s   
zBroadcastCursor.purgec                 C   s   | S r   r   r    r   r   r   Ú__iter__9   s   zBroadcastCursor.__iter__c              
   C   sd   	 zt | jƒ}W n tjjy' } zdt|ƒv r"|  ¡  W Y d }~q ‚ d }~ww 	 |  jd7  _|S )NTznot valid at serverr   )Únextr   Úpymongor   ÚOperationFailureÚstrr   r   )r   ÚmsgÚexcr   r   r   Ú__next__<   s   €ø
zBroadcastCursor.__next__N)T)Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r!   r#   r   r%   r,   r&   r   r   r   r   r   #   s    
r   c                       sT  e Zd ZdZdZi 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ejjd Z‡ fdd„Zdd„ Zdd„ Z‡ fdd„Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Z‡ fd d!„ZdEd#d$„Zd%d&„ Z d'd(„ Z!dEd)d*„Z"d+d,„ Z#d-d.„ Z$d/d0„ Z%e&d1d2„ ƒZ'e&d3d4„ ƒZ(e&d5d6„ ƒZ)e&d7d8„ ƒZ*e&d9d:„ ƒZ+d;d<„ Z,d=d>„ Z-d?d@„ Z.dAdB„ Z/dCdD„ Z0‡  Z1S )FÚChannelzMongoDB Channel.TFNi † z	127.0.0.1i‰i  Úkombu_defaultÚmessageszmessages.routingzmessages.broadcastzmessages.queues)Úconnect_timeoutÚsslÚttlÚcapped_queue_sizeÚdefault_hostnameÚdefault_portÚdefault_databaseÚmessages_collectionÚrouting_collectionÚbroadcast_collectionÚqueues_collectionÚcalc_queue_sizec                    s&   t t| ƒj|i |¤Ž i | _| j d S r   )Úsuperr1   r   Ú_broadcast_cursorsÚclient)r   ÚvargsÚkwargs©Ú	__class__r   r   r   r   s   
zChannel.__init__c                 K   s4   | j r| jjd|i|||  |d¡dœdd� d S d S )NÚ_idú	x-expires)rG   ÚoptionsÚ	expire_atT©Úupsert)r6   ÚqueuesÚupdateÚ_get_expire)r   ÚqueuerD   r   r   r   Ú
_new_queue|   s   
þ
ûÿzChannel._new_queuec                 C   s€   || j v rz	t|  |¡ƒ}W n ty   d }Y nw | jjd|idtjfgdd�}| jr1|  	|¡ |d u r8t
ƒ ‚tt|d ƒƒS )NrP   ÚpriorityT)ÚqueryÚsortÚremoveÚpayload)Ú_fanout_queuesr&   Ú_get_broadcast_cursorÚStopIterationr3   Úfind_and_modifyr'   Ú	ASCENDINGr6   Ú_update_queues_expirer	   r   r   )r   rP   r*   r   r   r   Ú_get…   s    
ÿ
ý
zChannel._getc                    sB   | j stt| ƒ |¡S || jv r|  |¡ ¡ S | j d|i¡ 	¡ S ©NrP   )
r?   r@   r1   Ú_sizerW   rX   r!   r3   Úfindr   ©r   rP   rE   r   r   r_   š   s
   
zChannel._sizec                 K   s@   t |ƒ|| j|dd�dœ}| jr|  |d¡|d< | j |¡ d S )NT)Úreverse)rV   rP   rR   zx-message-ttlrJ   )r   Ú_get_message_priorityr6   rO   r3   Úinsert)r   rP   ÚmessagerD   Údatar   r   r   Ú_put¥   s   ýzChannel._putc                 K   s   | j  t|ƒ|dœ¡ d S )N)rV   rP   )Ú	broadcastrd   r   )r   Úexchangere   Úrouting_keyrD   r   r   r   Ú_put_fanout±   s   ÿzChannel._put_fanoutc                 C   s:   |   |¡}|| jv r|  |¡ ¡  |S | j d|i¡ |S r^   )r_   rW   rX   r   r3   rU   )r   rP   Úsizer   r   r   Ú_purgeµ   s   

þzChannel._purgec                 C   s:   t | jj| d ƒ}| j d|i¡}|t dd„ |D ƒƒB S )NÚtableri   c                 s   s&   � | ]}|d  |d |d fV  qdS )rj   ÚpatternrP   Nr   )Ú.0Úrr   r   r   Ú	<genexpr>Å   s
   € ÿ
ÿz$Channel.get_table.<locals>.<genexpr>)Ú	frozensetÚstateÚ	exchangesÚroutingr`   )r   ri   ÚlocalRoutesÚbrokerRoutesr   r   r   Ú	get_table¿   s   ÿ

þzChannel.get_tablec                 C   sl   |   |¡jdkr|  ||||¡ || j|< ||||dœ}| ¡ }| jr+|  |d¡|d< | jj||dd� d S )NÚfanout)ri   rP   rj   ro   rH   rJ   TrK   )	ÚtypeofÚtypeÚ_create_broadcast_cursorrW   Úcopyr6   rO   rv   rN   )r   ri   rj   ro   rP   Úlookuprf   r   r   r   Ú_queue_bindÊ   s   ÿ
üzChannel._queue_bindc                    sŠ   | j  d|i¡ | jr| j d|i¡ tt| ƒj|fi |¤Ž || jv rCz| j 	|¡}W n
 t
y6   Y d S w | ¡  | j 	|¡ d S d S )NrP   rG   )rv   rU   r6   rM   r@   r1   Úqueue_deleterW   rA   ÚpopÚKeyErrorr#   )r   rP   rD   r   rE   r   r   r�   Þ   s   
ÿøzChannel.queue_deleteú
mongodb://c                 C   s  | j j}|j}| |¡s|| }|t|ƒd … s|| j7 }|jrBd|vrB| d¡\}}|j}|jr8|d|j 7 }|d | d | }|j	rH|j	n| j
}t ||¡}|d pW|j}	|	dv r_| j}	d| j| jrlt| jd ƒnd dœ}
|
 |d	 ¡ |  |
¡}
||	|
fS )
Nú@z://ú:Údatabase)ú/NTiè  )Úauto_start_requestr5   ÚconnectTimeoutMSrI   )Ú
connectionrB   ÚhostnameÚ
startswithÚlenr8   ÚuseridÚsplitÚpasswordÚportr9   r   Ú	parse_uriÚvirtual_hostr:   r5   r4   ÚintrN   Ú_prepare_client_options)r   ÚschemerB   rŒ   ÚheadÚtailÚcredentialsr’   ÚparsedÚdbnamerI   r   r   r   Ú
_parse_uriò   s4   

ÿü

zChannel._parse_uric                 C   sB   t jdkr| dd ¡ t| d¡tƒrt jj}||d  |d< |S )N©é   r‰   Úreadpreference)r'   Úversion_tupler‚   Ú
isinstanceÚgetr•   Úread_preferencesÚ_MONGOS_MODES)r   rI   Úmodesr   r   r   r–     s   
zChannel._prepare_client_optionsc                 K   s   t |fi |¤ŽS r   r   )r   Ú	argumentsrD   r   r   r   Úprepare_queue_arguments#  r"   zChannel.prepare_queue_argumentsc                 C   sÒ   | j |d�\}}}||d< tƒ }|dkrddlm} | ¡  n|dkr,ddlm} |ƒ  tdi |¤Ž}|| }	| ¡ d }
|
 	d	¡d }
t
tt|
 	d
¡ƒƒ}|dk rYtt |
¡ƒ‚| jrg|dk rgtt |
¡ƒ‚|	S )N)r—   ÚhostÚgeventr   )ÚmonkeyÚeventlet)Úmonkey_patchÚversionú-Ú.)r   rŸ   )é   r±   r   )r�   r   rª   r«   Ú	patch_allr¬   r­   r   Úserver_infor�   ÚtupleÚmapr•   r   ÚE_SERVER_VERSIONÚformatr6   ÚE_NO_TTL_INDEXES)r   r—   rŒ   rœ   ÚconfÚenvr«   r­   Ú	mongoconnr‡   Úversion_strr®   r   r   r   Ú_open&  s&   
zChannel._openc                 C   s*   | j | ¡ v r	dS |j| j | jdd� dS )z0Create capped collection for broadcast messages.NT)rl   Úcapped)r=   Úcollection_namesÚcreate_collectionr7   ©r   r‡   r   r   r   Ú_create_broadcastA  s   
þzChannel._create_broadcastc                 C   sŠ   || j  }|jg d¢dd� || j  dg¡ || j }| ddg¡ | jrC|jdgdd� |jdgdd� || j jdgdd� d	S d	S )
zEnsure indexes on collections.)©rP   r   )rR   r   )rG   r   T)Ú
backgroundrÃ   )ri   r   )rJ   r   r   )ÚexpireAfterSecondsN)r;   Úensure_indexr=   r<   r6   r>   )r   r‡   r3   rv   r   r   r   Ú_ensure_indexesJ  s   
ÿ


ÿüzChannel._ensure_indexesc                 C   s    |   ¡ }|  |¡ |  |¡ |S )zActually creates connection.)r½   rÂ   rÇ   rÁ   r   r   r   Ú_create_client]  s   

zChannel._create_clientc                 C   s   |   ¡ S r   )rÈ   r    r   r   r   rB   e  s   zChannel.clientc                 C   ó   | j | j S r   )rB   r;   r    r   r   r   r3   i  ó   zChannel.messagesc                 C   rÉ   r   )rB   r<   r    r   r   r   rv   m  rÊ   zChannel.routingc                 C   rÉ   r   )rB   r=   r    r   r   r   rh   q  rÊ   zChannel.broadcastc                 C   rÉ   r   )rB   r>   r    r   r   r   rM   u  rÊ   zChannel.queuesc              	   C   s6   z| j | W S  ty   |  | j| d d |¡ Y S w r   )rA   rƒ   r}   rW   ra   r   r   r   rX   y  s   ÿüzChannel._get_broadcast_cursorc                 C   sR   t jdkrd|itjdœ}nd|iddœ}| jjdi |¤Ž}t|ƒ }| j|< |S )Nrž   rP   )ÚfilterÚcursor_typeT)rS   Útailabler   )r'   r¡   r   ÚTAILABLErh   r`   r   rA   )r   ri   rj   ro   rP   rS   r   Úretr   r   r   r}   „  s   
þþz Channel._create_broadcast_cursorc              	   C   sn   t |tƒr| j d|i¡}|sdS |d }n|}z|d | }W n ttfy,   Y dS w |  ¡ tj|d� S )z”Get expiration header named `argument` of queue definition.

        Note:
            `queue` must be either queue name or options itself.
        rG   NrI   r§   )Úmilliseconds)	r¢   r
   rM   Úfind_onerƒ   Ú	TypeErrorÚget_nowÚdatetimeÚ	timedelta)r   rP   ÚargumentÚdocrf   Úvaluer   r   r   rO   ”  s   

ÿzChannel._get_expirec                 C   sT   |   |d¡}|s
dS | jjd|idd|iidd� | jjd|idd|iidd� dS )	z,Update expiration field on queues documents.rH   NrP   z$setrJ   T)ÚmultirG   )rO   rv   rN   rM   )r   rP   rJ   r   r   r   r\   «  s   ÿ
ÿzChannel._update_queues_expirec                 C   s
   t j  ¡ S )zReturn current time in UTC.)rÔ   Úutcnowr    r   r   r   rÓ   ·  s   
zChannel.get_now)r„   )2r-   r.   r/   r0   Úsupports_fanoutrW   r5   r6   r4   r7   r?   r8   r9   r:   r;   r<   r=   r>   r   r1   Úfrom_transport_optionsr   rQ   r]   r_   rg   rk   rm   ry   r€   r�   r�   r–   r¨   r½   rÂ   rÇ   rÈ   r   rB   r3   rv   rh   rM   rX   r}   rO   r\   rÓ   Ú__classcell__r   r   rE   r   r1   R   sb    
	

)
	




r1   c                   @   sn   e Zd ZdZeZdZdZejZej	j
ejf Z
ej	jejejf ZdZdZej	jjeg d¢ƒd�Zdd	„ Zd
S )Ú	TransportzMongoDB Transport.Tr   Úmongodbr'   )ÚdirectÚtopicrz   )Úexchange_typec                 C   s   t jS r   )r'   r®   r    r   r   r   Údriver_versionÓ  s   zTransport.driver_versionN)r-   r.   r/   r0   r1   Úcan_parse_urlÚpolling_intervalr9   r   rÞ   Úconnection_errorsr   ÚConnectionFailureÚchannel_errorsr(   Údriver_typeÚdriver_nameÚ
implementsÚextendrs   rã   r   r   r   r   rÞ   ¼  s&    ÿþÿ
ÿrÞ   )#r0   Ú
__future__r   r   rÔ   r'   r   r   r   Úpymongo.cursorr   Úkombu.exceptionsr   Ú
kombu.fiver	   r
   Úkombu.utils.compatr   Úkombu.utils.encodingr   Úkombu.utils.jsonr   r   Úkombu.utils.objectsr   Ú r   Úbaser   r¶   r¸   Úobjectr   r1   rÞ   r   r   r   r   Ú<module>   s,    /  l