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 closedr:rr#r$) memoryviewlensendBLOCKING_IO_ERRORSfilenorr.r create_future add_reader remove_readerr add_writer remove_writer_HAVE_PYOPENSSLr) rr!r4viewsentrAr5fdr:r6r6r7r0]sH           r0connrlengthintonceOptional[bool]rFc 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) Nrr:rr#r$cSr;r<r=r@r6r6r7rArBz,_async_socket_receive_ssl.._is_readyconnection closedrCrDrE)rF bytearray recv_intoOSErrorrIrJrr.rrKrLrMrrNrOrPr) rTrUr4rWmv total_readrAreadr5rSr:r6r6r7_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)NrrbrYrcrd) rFrZr[r\rIr+rerfrg)rTrUrarWr]r^rjr_r6r6r7r`s,  cCs||dSr<)sendall)rr!r6r6r7rkrkrcs" |jjrdSttIdHqr<)cancel_context cancelledr+re _POLL_TIMEOUTrTr6r6r7_poll_cancellations rqc s|}|}|dt}zIz5tr2t|ttfr2tjt |||dd|dIdHWW||Stjt ||||dIdHWW||Stj yY}zt d|d}~ww||w)Nr%T)rWr&r()r)r*r+r,r-r.rrr/r`_async_socket_receiver2r3r')rrU sock_timeoutr'r4errr6r6r7async_receive_data_sockets,      ru socket.socketcsXtt|}d}||kr*||||dIdH}|dkr"td||7}||ks |S)NrrY)rFrZsock_recv_intor\)rTrUr4r] bytes_read chunk_lengthr6r6r7rrs rrPyPyrdeadlineOptional[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.FrCNTpendingr)r_r'operation cancelledr()rTrrJhasattrr}time monotonicrfrgrosocket_checkerselectrmrnrr3r')rTr{r timed_outreadable remainingr'r6r6r7 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(rY)rZrFrTr)_PYPYrr get_timeoutset_conn_timeoutrfrrrgror[rIrmrnrr3r'r\rerrnoEINTR) rTrUr{r!r]rx orig_timeout short_timeoutryr5r6r6r7 receive_dataDsZ          & & rc@s^eZdZdddZedddZdd d Zdd dZdddZedddZ edddZ dS)NetworkingInterfaceBaserTr cC ||_dSr<rpselfrTr6r6r7__init__} z NetworkingInterfaceBase.__init__r#cCtr<NotImplementedErrorrr6r6r7r)z"NetworkingInterfaceBase.gettimeoutr' float | Noner$cCrr<rrr'r6r6r7r*z"NetworkingInterfaceBase.settimeoutcCrr<rrr6r6r7closerzNetworkingInterfaceBase.closeboolcCrr<rrr6r6r7 is_closingrz"NetworkingInterfaceBase.is_closingcCrr<rrr6r6r7get_connrz NetworkingInterfaceBase.get_conncCrr<rrr6r6r7rrzNetworkingInterfaceBase.sockN)rTr )r#r r'rr#r$r#r) __name__ __module__ __qualname__rpropertyr)r*rrrrr6r6r6r7r|s      rcsfeZdZdfdd ZedddZdd d Zdd dZdddZedddZ edddZ Z S)AsyncNetworkingInterfacerT!tuple[Transport, PyMongoProtocol]ct|dSr<superrr __class__r6r7rz!AsyncNetworkingInterface.__init__r#rcCs |jdjSNrTr)rr6r6r7r)s z#AsyncNetworkingInterface.gettimeoutr'r$cCs|jd|dSrrTr*rr6r6r7r*sz#AsyncNetworkingInterface.settimeoutcs(|jd|jdIdHdSr)rTr wait_closedrr6r6r7rszAsyncNetworkingInterface.closercCs|jdS)NrrTrrr6r6r7rrlz#AsyncNetworkingInterface.is_closingPyMongoProtocolcCs |jdSrrprr6r6r7rs z!AsyncNetworkingInterface.get_connrvcCs|jddS)Nrr3)rTget_extra_inforr6r6r7rszAsyncNetworkingInterface.sock)rTrr#rrr#r$r)r#r)r#rv) rrrrrr)r*rrrr __classcell__r6r6rr7rs     rcsveZdZdfdd ZdddZdd d Zd d dZd!ddZed"ddZ ed"ddZ d#ddZ d$ddZ Z S)%NetworkingInterfacerTr crr<rrrr6r7rrzNetworkingInterface.__init__r#rcC |jSr<rrr6r6r7r)rzNetworkingInterface.gettimeoutr'r$cCs|j|dSr<rrr6r6r7r*rzNetworkingInterface.settimeoutcCs|jdSr<)rTrrr6r6r7rrlzNetworkingInterface.closercCrr<rrr6r6r7rrzNetworkingInterface.is_closingcC|jSr<rprr6r6r7rzNetworkingInterface.get_conncCrr<rprr6r6r7rrzNetworkingInterface.sockrVcCrr<)rTrJrr6r6r7rJrzNetworkingInterface.filenobufferr"cCs |j|Sr<)rTr[)rrr6r6r7r[s zNetworkingInterface.recv_into)rTr rrrr)r#r )r#rV)rr"r#rV)rrrrr)r*rrrrrrJr[rr6r6rr7rs       rc@seZdZd3d4ddZd5d d Zed6d d Zd7ddZd8ddZd9ddZ d:ddZ d;d!d"Z dd+d,Zd3d>d-d.Zd3d>d/d0Zd?d1d2ZdS)@rNr'r|cCsd|_ttd|_d|_ttd|_d|_d|_d|_d|_ d|_ d|_ d|_ d|_ d|_||_d|_d|_t|_d|_t|_t|_t|_dS)Nr TF) transportrFrZ_header _header_index_compression_header_compression_index_message_message_index_expecting_header_expecting_compression _message_size_op_code_connection_lost _read_waiter_timeout_is_compressed_compressor_idr_max_message_size _response_tor+r,rK_closed collectionsdeque_pending_messages_done_messagesrr6r6r7rs* zPyMongoProtocol.__init__rr#r$cCrr<rrr6r6r7r*rzPyMongoProtocol.settimeoutcCr)zGThe configured timeout for the socket that underlies our protocol pair.rrr6r6r7r)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)rset_write_buffer_limitsr)rrr6r6r7connection_madeszPyMongoProtocol.connection_mademessager"cs.|jr td|j||jdS)z/Write a message to this connection's transport.zConnection is closedN)rrr\writeresume_reading)rrr6r6r7rs   zPyMongoProtocol.write request_id Optional[int]max_message_sizerVtuple[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 rY)rrAttributeErrorr\rrpopleftrr+r,rKrappendremoverr) rrrr read_waiterop_code compressor_id response_todatar6r6r7r_sB          zPyMongoProtocol.readsizehintrFcCs\|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) rrrFrZrrrrrrr)rrr6r6r7 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 datarrYNrFrT)rr\rrrprocess_headerrrrrrrFrZrrprocess_compression_headerrrr pause_readingrrr+r,rKr>r?rr)rrr5resultr6r6r7buffer_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)rrU_rrexpecting_compressionr6r6r7rss(   zPyMongoProtocol.process_headertuple[int, int]cCst|j\}}}||fS)z2Unpack a MongoDB Wire Protocol compression header.)_UNPACK_COMPRESSION_HEADERr)rrrrr6r6r7rsz*PyMongoProtocol.process_compression_headerr5Optional[Exception]cCsJt|j}|D]}|s|dur|dn|||j|qdSr<)listrr>r? set_exceptionrr)rr5r}msgr6r6r7_resolve_pending_messagess   z)PyMongoProtocol._resolve_pending_messagescCs|j||d|_dS)NT)rabortrrrr5r6r6r7rs   zPyMongoProtocol.closecCs(|||js|jddSdSr<)rrr>r?rr6r6r7connection_losts  zPyMongoProtocol.connection_lostcs|jIdHdSr<)rrr6r6r7rszPyMongoProtocol.wait_closedr<)r'r|rr)rrr#r$)rr"r#r$)rrrrVr#r)rrVr#rF)rrVr#r$)r#r)r#r)r5rr#r$r)rrrrr*rr)rrr_rrrrrrr rr6r6r6r7rs        "  ;   rc sLztj|||jdIdHWdStjy%}ztd|d}~ww)Nr&r()r+r/rr)r2r3r')rTr!r5r6r6r7 async_sendalls" r rrrUnion[_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' return_whenr( Got opcode rr~)rTr)r r get_deadlinerrrfrrqrr_r+waitFIRST_COMPLETEDcancelrGr3r'rrKeyErrorrkeysrCancelledError)rTrrr'r{cancellation_task read_tasktasksr>r}taskrr unpack_replyr6r6r7async_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) r Nrrrrrrrrrrr)r rrrTr)rrrrrrrrrr) rTrrr{r'rUrrrrrrr6r6r7receive_messagesD     r)rr r!r"r#r$)rr r!r"r4rr#r$)F) rTrrUrVr4rrWrXr#rF)rr r!r"rarr#r$) rTrrUrVrarrWrXr#rF)rTrr#r$)rr rUrVr#rF)rTrvrUrVr4rr#rF)rTrr{r|r#r$)rTrrUrVr{r|r#rF)rTrr!r"r#r$)rTrrrrrVr#r )rTrrrrrVr#r )P__doc__ __future__rr+rrr3structsysrrrrrrtypingrr r r pymongor r pymongo._asyncio_taskrpymongo.commonrpymongo.compression_supportrpymongo.errorsrrpymongo.messagerrrpymongo.socket_checkerrsslrrr- ImportErrorpymongo.pyopenssl_contextrrPpymongo.ssl_supportrrrpymongo.asynchronous.poolrpymongo.synchronous.poolrStructunpackrrroBlockingIOErrorrIr8platformr0r`rkrqrurrversionrrrrrrrr rrr6r6r6r7s               , 6      !8 X 4