o sDit@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:ejEvZFej>dkZGd]d>d?ZHd^d@dAZIGdBdCdCZJGdDdEdEeJZKGdFdGdGeJZLGdHdIdIe ZMd_dJdKZNefd`dPdQZOefdadRdSZPdS)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:r;r#r$) memoryviewlensendBLOCKING_IO_ERRORSfilenorr.r create_future add_reader remove_readerr add_writer remove_writer_HAVE_PYOPENSSLr) rr!r4viewsentrBr5fdr:r6r6r7r0]sH           r0connrlengthintonceOptional[bool]rGc 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:r;r#r$cSr<r=r>rAr6r6r7rBrCz,_async_socket_receive_ssl.._is_readyconnection closedrDrErF)rG bytearray recv_intoOSErrorrJrKrr.rrLrMrNrrOrPrQr) rUrVr4rXmv total_readrBreadr5rTr:r6r6r7_async_socket_receive_sslsT            %radummycst|}t|}d}d}||krMz |||d}Wnty/t|IdHd}Ynw|dkrt|IdHd}Ynw|dkrKt|dd}nt|dd}||7}||ks|S)NrrcrZrdre) rGr[r\r]rJr+rfrgrh)rUrVrbrXr^r_rkr`r6r6r7ras,  cCs||dSr=)sendall)rr!r6r6r7rlrlrcs" |jjrdSttIdHqr=)cancel_context cancelledr+rf _POLL_TIMEOUTrUr6r6r7_poll_cancellations rrc s|}|}|dt}zIz5tr2t|ttfr2tjt |||dd|dIdHWW||Stjt ||||dIdHWW||Stj yY}zt d|d}~ww||w)Nr%T)rXr&r()r)r*r+r,r-r.rrr/ra_async_socket_receiver2r3r')rrV sock_timeoutr'r4errr6r6r7async_receive_data_sockets,      rv socket.socketcsXtt|}d}||kr*||||dIdH}|dkr"td||7}||ks |S)NrrZ)rGr[sock_recv_intor])rUrVr4r^ bytes_read chunk_lengthr6r6r7rss rsPyPyrdeadlineOptional[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.FrDNTpendingr)r`r'operation cancelledr()rUrrKhasattrr~time monotonicrgrhrpsocket_checkerselectrnrorr3r')rUr|r timed_outreadable remainingr'r6r6r7 wait_for_read$s*   rc Cst|}t|}d}|j}z||krzBtstr3t||tr2|dur2| t |t dn|durDt t |t dt}nt}| ||j||d}Wn^tyl|jjrftddtddtjy|jjr|tddtsts|js|dur|t dkrYqty} z|jjrtddt| tjkrWYd} ~ qd} ~ ww|dkrtd||7}||ksW| ||SW| ||S| |w)Nrrr(rZ)r[rGrUr)_PYPY_WINDOWSrr get_timeoutset_conn_timeoutrgrrrhrpr\rJrnrorr3r'is_sdamr]rerrnoEINTR) rUrVr|r!r^ry orig_timeout short_timeoutrzr5r6r6r7 receive_dataEsh          - - rc@s^eZdZdddZedddZdd d Zdd dZdddZedddZ edddZ dS)NetworkingInterfaceBaserUr cC ||_dSr=rqselfrUr6r6r7__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)rUr )r#r r'rr#r$r#r) __name__ __module__ __qualname__rpropertyr)r*rrrrr6r6r6r7rs      rcsfeZdZdfdd ZedddZdd d Zdd dZdddZedddZ edddZ Z S)AsyncNetworkingInterfacerU!tuple[Transport, PyMongoProtocol]ct|dSr=superrr __class__r6r7rz!AsyncNetworkingInterface.__init__r#rcCs |jdjSNrUr)rr6r6r7r)s z#AsyncNetworkingInterface.gettimeoutr'r$cCs|jd|dSrrUr*rr6r6r7r*sz#AsyncNetworkingInterface.settimeoutcs(|jd|jdIdHdSr)rUr wait_closedrr6r6r7rszAsyncNetworkingInterface.closercCs|jdS)NrrUrrr6r6r7rrmz#AsyncNetworkingInterface.is_closingPyMongoProtocolcCs |jdSrrqrr6r6r7rs z!AsyncNetworkingInterface.get_connrwcCs|jddS)Nrr3)rUget_extra_inforr6r6r7rszAsyncNetworkingInterface.sock)rUrr#rrr#r$r)r#r)r#rw) rrrrrr)r*rrrr __classcell__r6r6rr7rs     rcsveZdZdfdd ZdddZdd d Zd d dZd!ddZed"ddZ ed"ddZ d#ddZ d$ddZ Z S)%NetworkingInterfacerUr crr=rrrr6r7rrzNetworkingInterface.__init__r#rcC |jSr=rrr6r6r7r)rzNetworkingInterface.gettimeoutr'r$cCs|j|dSr=rrr6r6r7r*rzNetworkingInterface.settimeoutcCs|jdSr=)rUrrr6r6r7rrmzNetworkingInterface.closercCrr=rrr6r6r7rrzNetworkingInterface.is_closingcC|jSr=rqrr6r6r7rzNetworkingInterface.get_conncCrr=rqrr6r6r7rrzNetworkingInterface.sockrWcCrr=)rUrKrr6r6r7rKrzNetworkingInterface.filenobufferbytes | memoryviewcCs |j|Sr=)rUr\)rrr6r6r7r\s zNetworkingInterface.recv_into)rUr rrrr)r#r )r#rW)rrr#rW)rrrrr)r*rrrrrrKr\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) transportrGr[_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,rL_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_sizerWtuple[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 rZ)rrAttributeErrorr]rrpopleftrr+r,rLrappendremoverr) rrrr read_waiterop_code compressor_id response_todatar6r6r7r` sB          zPyMongoProtocol.readsizehintrGcCs\|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) rrrGr[rrrrrrr)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 datarrZNrFrT)rr]rrrprocess_headerrrrrrrGr[rrprocess_compression_headerrrr pause_readingrrr+r,rLr?r@rr)rrr5resultr6r6r7buffer_updated@sh          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)rrV_rrexpecting_compressionr6r6r7r{s(   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)rabortr rrr5r6r6r7rs   zPyMongoProtocol.closecCs(|||js|jddSdSr=)r rr?r@r r6r6r7connection_losts  zPyMongoProtocol.connection_lostcs|jIdHdSr=)rrr6r6r7rszPyMongoProtocol.wait_closedr=)r'r}rr)rrr#r$)rr"r#r$)rrrrWr#r)rrWr#rG)rrWr#r$)r#r)r#r)r5rr#r$r)rrrrr*rr)rrr`rrrrr rr rr6r6r6r7rs        "  ;   rc sLztj|||jdIdHWdStjy%}ztd|d}~ww)Nr&r()r+r/rr)r2r3r')rUr!r5r6r6r7 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' return_whenr( Got opcode rr)rUr)r r get_deadlinerrrgrrrrr`r+waitFIRST_COMPLETEDcancelrHr3r'rrKeyErrorrkeysrCancelledError)rUrrr'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) rNrrrrrrrrrrr)r rrrUr)rrrrrrrrrr) rUrrr|r'rVrrrrrrr6r6r7receive_messagesD     r )rr r!r"r#r$)rr r!r"r4rr#r$)F) rUrrVrWr4rrXrYr#rG)rr r!r"rbrr#r$) rUrrVrWrbrrXrYr#rG)rUrr#r$)rr rVrWr#rG)rUrwrVrWr4rr#rG)rUrr|r}r#r$)rUrrVrWr|r}r#rG)rUrr!r"r#r$)rUrrrrrWr#r)rUrrrrrWr#r)Q__doc__ __future__rr+rrr3structsysrrrrrrtypingrr r r pymongor r pymongo._asyncio_taskrpymongo.commonrpymongo.compression_supportrpymongo.errorsrrpymongo.messagerrrpymongo.socket_checkerrsslrrr- ImportErrorpymongo.pyopenssl_contextrrQpymongo.ssl_supportrrrpymongo.asynchronous.poolrpymongo.synchronous.poolrStructunpackrrrpBlockingIOErrorrJr8platformr0rarlrrrvrsversionrrrrrrrrrrr r6r6r6r7s               , 6       !? X 4