o hbs@sdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddlm Z m Z m Z m Z mZddlmZmZmZmZddlmZmZddlmZddlmZdd lmZdd lmZmZdd l m!Z!m"Z"m#Z#dd l$m%Z%z dd l&m'Z'm(Z(dZ)Wn e*ydZ)Ynwz ddl+m,Z,dZ-Wn e*ydZ-e(Z,Ynwddl.m/Z/m0Z0m1Z1erddl2m3Z3ddl4m5Z5e6dj7Z8e6dj7Z9dZ:e;ge/ej<RZdkrdUd"d#Z? dVdWd+d,Z@n dXd.d#Z? dVdYd/d,Z@dTd0d1ZAdZd3d4ZBd[d5d6ZCd\d8d9ZDd:ejEvZFd]d>d?ZGd^d@dAZHGdBdCdCZIGdDdEdEeIZJGdFdGdGeIZKGdHdIdIe ZLd_dJdKZMefd`dPdQZNefdadRdSZOdS)bz&Internal network layer helper methods.) annotationsN)AbstractEventLoop BaseTransportBufferedProtocolFuture Transport) TYPE_CHECKINGAnyOptionalUnion)_csot ssl_support) create_task)MAX_MESSAGE_SIZE) decompress) ProtocolError_OperationCancelled) _UNPACK_REPLY_OpMsg_OpReply)_errno_from_exception)SSLError SSLSocketTF)_sslConn)BLOCKING_IO_LOOKUP_ERRORBLOCKING_IO_READ_ERRORBLOCKING_IO_WRITE_ERROR)AsyncConnection) Connectionz._is_ready!Underlying socket has been closedr9rr#r$) memoryviewlensendBLOCKING_IO_ERRORSfilenorr.r create_future add_reader remove_readerr add_writer remove_writer_HAVE_PYOPENSSLr) rr!r3viewsentr@r4fdr9r5r5r6r0]sH           r0connrlengthintonceOptional[bool]rEc stt|}d}d dd}||krz |||d}|dkr$td|r-|d|WS||7}Wnty}z|} | d krGtd dt|trj| } | | || z | IdHW| | n| | wt|t r| } | | || z | IdHW|| n|| wtrt|tr| } | | || z| | || | IdHW| | || n | | || wWYd}~nd}~ww||ks|S) Nrr9rr#r$cSr:r;r<r?r5r5r6r@rAz,_async_socket_receive_ssl.._is_readyconnection closedrBrCrD)rE bytearray recv_intoOSErrorrHrIrr.rrJrKrLrrMrNrOr) rSrTr3rVmv total_readr@readr4rRr9r5r5r6_async_socket_receive_sslsT            %r_dummycst|}t|}d}d}||krMz |||d}Wnty/t|IdHd}Ynw|dkrt|IdHd}Ynw|dkrKt|dd}nt|dd}||7}||ks|S)NrrarXrbrc) rErYrZr[rHr+rdrerf)rSrTr`rVr\r]rir^r5r5r6r_s,  cCs||dSr;)sendall)rr!r5r5r6rjrjrcs" |jjrdSttIdHqr;)cancel_context cancelledr+rd _POLL_TIMEOUTrSr5r5r6_poll_cancellations rpc s|}|}|dt}zIz5tr2t|ttfr2tjt |||dd|dIdHWW||Stjt ||||dIdHWW||Stj yY}zt d|d}~ww||w)Nr%T)rVr&r()r)r*r+r,r-r.rrr/r__async_socket_receiver1r2r')rrT sock_timeoutr'r3errr5r5r6async_receive_data_sockets,      rt socket.socketcsXtt|}d}||kr*||||dIdH}|dkr"td||7}||ks |S)NrrX)rErYZsock_recv_intor[)rSrTr3r\ bytes_read chunk_lengthr5r5r6rqs rqPyPyrdeadlineOptional[float]cCs|jj}d}|dkrdS t|dr|dkrd}n"|r4|t}|dkr+d}tt|t d}nt }|j j |d|d}|j j rGtd|rKdS|rRtd q) zABlock until at least one byte is read, or a timeout, or a cancel.FrBNTpendingr)r^r'operation cancelledr()rSrrIhasattrr{time monotonicrerfrnZsocket_checkerselectrlrmrr2r')rSryrZ timed_outreadable remainingr'r5r5r6 wait_for_read#s*   rc Cst|}t|}d}|j}z||krz@tr1t||tr0|dur0|t |t dn|durBt t |t dt }nt }|||j||d}WnMtyj|jjrdtddtddtjy|jjrztddtr}Yqty} z|jjrtddt| tjkrWYd} ~ qd} ~ ww|dkrtd||7}||ksW|||SW|||S||w)Nrr|r(rX)rYrErSr)_PYPYrr get_timeoutZset_conn_timeoutrer~rrfrnrZrHrlrmrr2r'r[rerrnoZEINTR) rSrTryr!r\rv orig_timeoutZ short_timeoutrwr4r5r5r6 receive_dataDsZ          & & rc@s^eZdZdddZedddZdd d Zdd dZdddZedddZ edddZ dS)NetworkingInterfaceBaserSr cC ||_dSr;roselfrSr5r5r6__init__} z NetworkingInterfaceBase.__init__r#cCtr;NotImplementedErrorrr5r5r6r)z"NetworkingInterfaceBase.gettimeoutr' float | Noner$cCrr;rrr'r5r5r6r*z"NetworkingInterfaceBase.settimeoutcCrr;rrr5r5r6closerzNetworkingInterfaceBase.closeboolcCrr;rrr5r5r6 is_closingrz"NetworkingInterfaceBase.is_closingcCrr;rrr5r5r6get_connrz NetworkingInterfaceBase.get_conncCrr;rrr5r5r6rrzNetworkingInterfaceBase.sockN)rSr )r#r r'rr#r$r#r) __name__ __module__ __qualname__rpropertyr)r*rrrrr5r5r5r6r|s      rcsfeZdZdfdd ZedddZdd d Zdd dZdddZedddZ edddZ Z S)AsyncNetworkingInterfacerS!tuple[Transport, PyMongoProtocol]ct|dSr;superrr __class__r5r6rz!AsyncNetworkingInterface.__init__r#rcCs |jdjSNrSr)rr5r5r6r)s z#AsyncNetworkingInterface.gettimeoutr'r$cCs|jd|dSrrSr*rr5r5r6r*sz#AsyncNetworkingInterface.settimeoutcs(|jd|jdIdHdSr)rSr wait_closedrr5r5r6rszAsyncNetworkingInterface.closercCs|jdS)NrrSrrr5r5r6rrkz#AsyncNetworkingInterface.is_closingPyMongoProtocolcCs |jdSrrorr5r5r6rs z!AsyncNetworkingInterface.get_connrucCs|jddS)Nrr2)rSZget_extra_inforr5r5r6rszAsyncNetworkingInterface.sock)rSrr#rrr#r$r)r#r)r#ru) rrrrrr)r*rrrr __classcell__r5r5rr6rs     rcsveZdZdfdd ZdddZdd d Zd d dZd!ddZed"ddZ ed"ddZ d#ddZ d$ddZ Z S)%NetworkingInterfacerSr crr;rrrr5r6rrzNetworkingInterface.__init__r#rcC |jSr;rrr5r5r6r)rzNetworkingInterface.gettimeoutr'r$cCs|j|dSr;rrr5r5r6r*rzNetworkingInterface.settimeoutcCs|jdSr;)rSrrr5r5r6rrkzNetworkingInterface.closercCrr;rrr5r5r6rrzNetworkingInterface.is_closingcC|jSr;rorr5r5r6rzNetworkingInterface.get_conncCrr;rorr5r5r6rrzNetworkingInterface.sockrUcCrr;)rSrIrr5r5r6rIrzNetworkingInterface.filenobufferr"cCs |j|Sr;)rSrZ)rrr5r5r6rZs zNetworkingInterface.recv_into)rSr rrrr)r#r )r#rU)rr"r#rU)rrrrr)r*rrrrrrIrZrr5r5rr6rs       rc@seZdZd3d4ddZd5d d Zed6d d Zd7ddZd8ddZd9ddZ d:ddZ d;d!d"Z dd+d,Zd3d>d-d.Zd3d>d/d0Zd?d1d2ZdS)@rNr'rzcCsd|_ttd|_d|_ttd|_d|_d|_d|_d|_ d|_ d|_ d|_ d|_ d|_||_d|_d|_t|_d|_t|_t|_t|_dS)Nr TF) transportrErY_header _header_index_compression_header_compression_index_message_message_index_expecting_header_expecting_compression _message_size_op_code_connection_lostZ _read_waiter_timeoutZ_is_compressed_compressor_idr_max_message_size _response_tor+r,rJ_closed collectionsdeque_pending_messages_done_messagesrr5r5r6rs* zPyMongoProtocol.__init__rr#r$cCrr;rrr5r5r6r*rzPyMongoProtocol.settimeoutcCr)zGThe configured timeout for the socket that underlies our protocol pair.rrr5r5r6r)szPyMongoProtocol.gettimeoutrrcCs||_|jttdS)zCalled exactly once when a connection is made. The transport argument is the transport representing the write side of the connection. N)rZset_write_buffer_limitsr)rrr5r5r6connection_madeszPyMongoProtocol.connection_mademessager"cs.|jr td|j||jdS)z/Write a message to this connection's transport.zConnection is closedN)rrr[writeresume_reading)rrr5r5r6rs   zPyMongoProtocol.write request_id Optional[int]max_message_sizerUtuple[bytes, int]c s|jrz|jWn tytddw||_|jr'|jIdH}n8|jr3|jr3tdt }|j |z|IdH}W||jvrQ|j |n ||jvr^|j |ww|r|\}}}}|dury||krytd|d||durt||}||fStd)zARead a single MongoDB Wire Protocol message from this connection.zconnection is already closedNGot response id  but expected rX)rrAttributeErrorr[rrpopleftrr+r,rJrappendremoverr) rrrrZ read_waiterop_code compressor_id response_todatar5r5r6r^sB          zPyMongoProtocol.readsizehintrEcCs\|jr|js ttd|_|jS|jr|j|jdS|jr&|j|j dS|j|j dS)a8Called to allocate a new receive buffer. The asyncio loop calls this method expecting to receive a non-empty buffer to fill with data. If any data does not fit into the returned buffer, this method will be called again until either no data remains or an empty buffer is returned. i@N) rrrErYrrrrrrr)rrr5r5r6 get_buffer#s zPyMongoProtocol.get_buffernbytesc Cs|dkr |tddS|jrdS|jrU|j|7_|jdkrSd|_z|\|_|_|_|_ Wnt yJ}z ||WYd}~dSd}~wwt t |j|_ dS|j rq|j|7_|jdkrod|_ |\|_|_dS|j|7_|j|jkrd|_|j|jr|j}nt}|r|ddS||j|j|j|j f|j|d|_d|_d|_d|_d|_ d|_d|_d|_dSdS)z9Called when the buffer was updated with the received datarrXNrFrT)rr[rrrprocess_headerrrrrrrErYrrprocess_compression_headerrrrZ pause_readingrrr+r,rJr=r>rr)rrr4resultr5r5r6buffer_updated8sh          zPyMongoProtocol.buffer_updatedtuple[int, int, int, bool]cCst|j\}}}}d}|dkr!|dkrtd|dd}|d8}|dkr-td|d ||jkr>td|d |jd |d|||fS) z&Unpack a MongoDB Wire Protocol header.FMessage length (zA) not longer than standard OP_COMPRESSED message header size (25)Trr3) not longer than standard message header size (16)*) is larger than server max message size ())_UNPACK_HEADERrrr)rrT_rrZexpecting_compressionr5r5r6rss(   zPyMongoProtocol.process_headertuple[int, int]cCst|j\}}}||fS)z2Unpack a MongoDB Wire Protocol compression header.)_UNPACK_COMPRESSION_HEADERr)rrrrr5r5r6rsz*PyMongoProtocol.process_compression_headerr4Optional[Exception]cCsJt|j}|D]}|s|dur|dn|||j|qdSr;)listrr=r>Z set_exceptionrr)rr4r{msgr5r5r6_resolve_pending_messagess   z)PyMongoProtocol._resolve_pending_messagescCs|j||d|_dS)NT)rabortrrrr4r5r5r6rs   zPyMongoProtocol.closecCs(|||js|jddSdSr;)rrr=r>rr5r5r6connection_losts  zPyMongoProtocol.connection_lostcs|jIdHdSr;)rrr5r5r6rszPyMongoProtocol.wait_closedr;)r'rzrr)rrr#r$)rr"r#r$)rrrrUr#r)rrUr#rE)rrUr#r$)r#r)r#r)r4rr#r$r)rrrrr*rr)rrr^rrrrrrrrr5r5r5r6rs        "  ;   rc sLztj|||jdIdHWdStjy%}ztd|d}~ww)Nr&r()r+r/rr)r1r2r')rSr!r4r5r5r6 async_sendalls" rrrrUnion[_OpReply, _OpMsg]c sZ|jj}trt}n |rt|}nd}|r$t|td}tt |}t|jj ||}||g}zZt j ||t jdIdH\}} | D]} | qI| rZt | IdHt|dkretd||vr|\} } zt| } Wntytd| dtdw| | WStdt jy|D]} | qt |IdHw)1Receive a raw BSON message or raise socket.error.Nr)r'Z return_whenr( Got opcode rr|)rSr)r r get_deadliner~rrerrprr^r+waitZFIRST_COMPLETEDcancelrFr2r'rrKeyErrorrkeysrZCancelledError)rSrrr'ryZcancellation_taskZ read_tasktasksr=r{taskrr unpack_replyr5r5r6async_receive_messagesR          rc Cs&tr t}n|j}|rt|}nd}tt|d|\}}}}|dur7||kr7t d|d||dkrCt d|d||krRt d|d|d|d krmt t|d |\}}} t t||d || } nt||d|} z t |} W| | St yt d |dt dw) rNrrrrrrrrrrr)r rrrSr)r~rrrrrrrrr) rSrrryr'rTrrrrrrr5r5r6receive_messagesD     r)rr r!r"r#r$)rr r!r"r3rr#r$)F) rSrrTrUr3rrVrWr#rE)rr r!r"r`rr#r$) rSrrTrUr`rrVrWr#rE)rSrr#r$)rr rTrUr#rE)rSrurTrUr3rr#rE)rSrryrzr#r$)rSrrTrUryrzr#rE)rSrr!r"r#r$)rSrrrrrUr#r)rSrrrrrUr#r)P__doc__ __future__rr+rrr2structsysr~rrrrrtypingrr r r Zpymongor r Zpymongo._asyncio_taskrZpymongo.commonrZpymongo.compression_supportrZpymongo.errorsrrZpymongo.messagerrrZpymongo.socket_checkerrsslrrr- ImportErrorZpymongo.pyopenssl_contextrrOZpymongo.ssl_supportrrrZpymongo.asynchronous.poolrZpymongo.synchronous.poolrStructunpackrrrnBlockingIOErrorrHr7platformr0r_rjrprtrqversionrrrrrrrrrrr5r5r5r6s               , 6      !8 X 4