o
    —¨Êh=  ã                   @  sî   d Z ddlmZ ddlZddlmZ ddl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 dd	lmZ dd
lmZmZ ddlmZ ddlmZ ddlmZ ddlmZ dZdZG dd„ dƒZG dd„ dej ƒZ G dd„ dej!ƒZ!dS )a  MongoDB transport module for kombu.

Features
========
* Type: Virtual
* Supports Direct: Yes
* Supports Topic: Yes
* Supports Fanout: Yes
* Supports Priority: Yes
* Supports TTL: Yes

Connection String
=================
 *Unreviewed*

Transport Options
=================

* ``connect_timeout``,
* ``ssl``,
* ``ttl``,
* ``capped_queue_size``,
* ``default_hostname``,
* ``default_port``,
* ``default_database``,
* ``messages_collection``,
* ``routing_collection``,
* ``broadcast_collection``,
* ``queues_collection``,
* ``calc_queue_size``,
é    )ÚannotationsN)ÚEmpty)ÚMongoClientÚerrorsÚ
uri_parser)Ú
CursorType)ÚVersionMismatch)Ú_detect_environment)Úbytes_to_str)ÚdumpsÚloads)Úcached_property)Úmaybe_sanitize_urlé   )Ú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   || _ d| _| jdd� d S )Nr   F)Úrewind)Ú_cursorÚ_offsetÚpurge)ÚselfÚcursor© r   úI/var/www/html/env/lib/python3.10/site-packages/kombu/transport/mongodb.pyÚ__init__C   s   zBroadcastCursor.__init__c                 C  s   | j j i ¡| j S ©N)r   Ú
collectionÚcount_documentsr   ©r   r   r   r   Úget_sizeH   s   zBroadcastCursor.get_sizec                 C  s   | j  ¡  d S r   )r   Úcloser    r   r   r   r"   K   s   zBroadcastCursor.closeTc                 C  s2   |r| j  ¡  | j j i ¡| _| j  | j¡| _ d S r   )r   r   r   r   r   Úskip)r   r   r   r   r   r   N   s   
zBroadcastCursor.purgec                 C  s   | S r   r   r    r   r   r   Ú__iter__V   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__Y   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                      s\  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dGd#d$„Zd%d&„ Z d'd(„ Z!dGd)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„ Z0dEdF„ Z1‡  Z2S )HÚ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 ƒ j|i |¤Ž i | _| j d S r   )Úsuperr   Ú_broadcast_cursorsÚclient)r   ÚvargsÚkwargs©Ú	__class__r   r   r   �   s   
zChannel.__init__c              	   K  s8   | j r| jjd|id|||  |d¡dœidd� d S d S )NÚ_idú$setú	x-expires)rF   ÚoptionsÚ	expire_atT©Úupsert)r5   ÚqueuesÚ
update_oneÚ_get_queue_expire)r   ÚqueuerC   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�}| jr0|  	|¡ |d u r7t
ƒ ‚tt|d ƒƒS )NrP   Úpriority)ÚsortÚpayload)Ú_fanout_queuesr%   Ú_get_broadcast_cursorÚStopIterationr2   Úfind_one_and_deleter&   Ú	ASCENDINGr5   Ú_update_queues_expirer   r   r
   )r   rP   r)   r   r   r   Ú_get¨   s   
ÿ
þ
zChannel._getc                   s:   | j s	tƒ  |¡S || jv r|  |¡ ¡ S | j d|i¡S ©NrP   )r>   r?   Ú_sizerU   rV   r!   r2   r   ©r   rP   rD   r   r   r]   ¼   s
   
zChannel._sizec                 K  sr   t |ƒ|| j|dd�dœ}| jr1|  |d¡|d< |  |¡}|d ur1|d d u s-||d k r1||d< | j |¡ d S )NT)Úreverse)rT   rP   rR   zx-message-ttlrJ   )r   Ú_get_message_priorityr5   rO   Ú_get_message_expirer2   Ú
insert_one)r   rP   ÚmessagerC   ÚdataÚ
msg_expirer   r   r   Ú_putÇ   s   ý
zChannel._putc                 K  s   | j  t|ƒ|dœ¡ d S )N)rT   rP   )Ú	broadcastrb   r   )r   Úexchangerc   Úrouting_keyrC   r   r   r   Ú_put_fanoutØ   s   ÿzChannel._put_fanoutc                 C  s:   |   |¡}|| jv r|  |¡ ¡  |S | j d|i¡ |S r\   )r]   rU   rV   r   r2   Údelete_many)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Útablerh   c                 s  s&   � | ]}|d  |d |d fV  qdS )ri   ÚpatternrP   Nr   )Ú.0Úrr   r   r   Ú	<genexpr>ì   s
   € ÿ
ÿz$Channel.get_table.<locals>.<genexpr>)Ú	frozensetÚstateÚ	exchangesÚroutingÚfind)r   rh   ÚlocalRoutesÚbrokerRoutesr   r   r   Ú	get_tableæ   s   ÿ

þzChannel.get_tablec                 C  sp   |   |¡jdkr|  ||||¡ || j|< ||||dœ}| ¡ }| jr+|  |d¡|d< | jj|d|idd� d S )NÚfanout)rh   rP   ri   ro   rH   rJ   rG   TrK   )	ÚtypeofÚtypeÚ_create_broadcast_cursorrU   Úcopyr5   rO   rv   rN   )r   rh   ri   ro   rP   Úlookuprd   r   r   r   Ú_queue_bindñ   s   ÿ
üzChannel._queue_bindc                   s†   | j  d|i¡ | jr| j d|i¡ tƒ j|fi |¤Ž || jv rAz| j 	|¡}W n
 t
y4   Y d S w | ¡  | j 	|¡ d S d S )NrP   rF   )rv   rk   r5   rM   Ú
delete_oner?   Úqueue_deleterU   r@   ÚpopÚKeyErrorr"   )r   rP   rC   r   rD   r   r   rƒ     s   
ÿøzChannel.queue_deleteú
mongodb://c                 C  s*  | j j}|j}| d¡rd}d| }| |¡s|| }|t|ƒd … s(|| j7 }|jrMd|vrM| d¡\}}|j}|jrC|d|j 7 }|d | d | }|j	rS|j	n| j
}t ||¡}|d pb|j}	|	dv rj| j}	d	| j| jrwt| jd
 ƒnd dœ}
|
 |d ¡ |  |
¡}
d|
v r�|
 d¡ ||	|
fS )Nzsrv://zmongodb+srv://zmongodb+ú@z://ú:Údatabase)ú/NTiè  )Úauto_start_requestr4   ÚconnectTimeoutMSrI   Útlsr4   )Ú
connectionrA   ÚhostnameÚ
startswithÚlenr7   ÚuseridÚsplitÚpasswordÚportr8   r   Ú	parse_uriÚvirtual_hostr9   r4   r3   ÚintÚupdateÚ_prepare_client_optionsr„   )r   ÚschemerA   r�   ÚheadÚtailÚcredentialsr•   ÚparsedÚdbnamerI   r   r   r   Ú
_parse_uri  s>   


ÿü


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š   I  s   
zChannel._prepare_client_optionsc                 K  s   t |fi |¤ŽS r   r   )r   Ú	argumentsrC   r   r   r   Úprepare_queue_argumentsQ  s   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Úformatr5   ÚE_NO_TTL_INDEXES)r   r›   r�   r    ÚconfÚenvr¯   r±   Ú	mongoconnr‰   Úversion_strr²   r   r   r   Ú_openT  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<   Úlist_collection_namesÚcreate_collectionr6   ©r   r‰   r   r   r   Ú_create_broadcasto  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   )rF   r   T)Ú
backgroundrÇ   )rh   r   )rJ   r   r   )ÚexpireAfterSecondsN)r:   Úcreate_indexr<   r;   r5   r=   )r   r‰   r2   rv   r   r   r   Ú_ensure_indexesx  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   rA   “  s   zChannel.clientc                 C  ó   | j | j S r   )rA   r:   r    r   r   r   r2   —  ó   zChannel.messagesc                 C  rÍ   r   )rA   r;   r    r   r   r   rv   ›  rÎ   zChannel.routingc                 C  rÍ   r   )rA   r<   r    r   r   r   rg   Ÿ  rÎ   zChannel.broadcastc                 C  rÍ   r   )rA   r=   r    r   r   r   rM   £  rÎ   zChannel.queuesc              	   C  s6   z| j | W S  ty   |  | j| d d |¡ Y S w r   )r@   r…   r~   rU   r^   r   r   r   rV   §  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)ÚqueryÚtailabler   )r&   r¥   r   ÚTAILABLErg   rw   r   r@   )r   rh   ri   ro   rP   rÑ   r   Úretr   r   r   r~   ²  s   
þþz Channel._create_broadcast_cursorc                 C  s6   |  di ¡  d¡}|d ur|  ¡ tjt|ƒd� S d S )NÚ
propertiesÚ
expiration©Úmilliseconds)r§   Úget_nowÚdatetimeÚ	timedeltar˜   )r   rc   Úvaluer   r   r   ra   Â  s   ÿzChannel._get_message_expirec              	   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.
        rF   NrI   r«   r×   )	r¦   r(   rM   Úfind_oner…   Ú	TypeErrorrÙ   rÚ   rÛ   )r   rP   ÚargumentÚdocrd   rÜ   r   r   r   rO   Ç  s   

ÿzChannel._get_queue_expirec                 C  sL   |   |d¡}|s
dS | j d|idd|ii¡ | j d|idd|ii¡ dS )z,Update expiration field on queues documents.rH   NrP   rG   rJ   rF   )rO   rv   Úupdate_manyrM   )r   rP   rJ   r   r   r   rZ   ß  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†   )3r,   r-   r.   r/   Úsupports_fanoutrU   r4   r5   r3   r6   r>   r7   r8   r9   r:   r;   r<   r=   r   r0   Úfrom_transport_optionsr   rQ   r[   r]   rf   rj   rm   rz   r�   rƒ   r¡   rš   r¬   rÁ   rÆ   rË   rÌ   r   rA   r2   rv   rg   rM   rV   r~   ra   rO   rZ   rÙ   Ú__classcell__r   r   rD   r   r0   o   sd    


0
	




r0   c                   @  sz   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ddd„ZdS )Ú	TransportzMongoDB Transport.Tr   Úmongodbr&   )ÚdirectÚtopicr{   )Úexchange_typec                 C  s   t jS r   )r&   r²   r    r   r   r   Údriver_version  s   zTransport.driver_versionFú**Úurir(   Úreturnc                 C  sB   |sdS |r|S d|vrt |ƒS | dd¡\}}d t |ƒ|g¡S )Nr†   ú,r   )r   r“   Újoin)r   rí   Úinclude_passwordÚmaskÚuri1Ú	remainderr   r   r   Úas_uri
  s   zTransport.as_uriN)Frì   )rí   r(   rî   r(   )r,   r-   r.   r/   r0   Úcan_parse_urlÚpolling_intervalr8   r   ræ   Úconnection_errorsr   ÚConnectionFailureÚchannel_errorsr'   Údriver_typeÚdriver_nameÚ
implementsÚextendrs   rë   rõ   r   r   r   r   ræ   ð  s(    ÿþÿ
ÿræ   )"r/   Ú
__future__r   rÚ   rP   r   r&   r   r   r   Úpymongo.cursorr   Úkombu.exceptionsr   Úkombu.utils.compatr	   Úkombu.utils.encodingr
   Úkombu.utils.jsonr   r   Úkombu.utils.objectsr   Úkombu.utils.urlr   Ú r   Úbaser   rº   r¼   r   r0   ræ   r   r   r   r   Ú<module>   s.    /   