o )fR@s$UddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z m Z ddl mZddlmZmZmZmZmZddlmZdZe dZGd d d eZGd d d eeZGd ddZGdddZGdddeZGdddeZeZ eee!d<GdddeeZ"Gddde"eZ#dS)N) AwaitableCallableDequeFinalGenericListOptionalTupleTypeVar) BaseProtocol) _EXC_SENTINELBaseTimerContext TimerNoop set_exception set_result)internal_logger) EMPTY_PAYLOAD EofStream StreamReader DataQueueFlowControlDataQueue_Tc@seZdZdZdS)rzeof stream indication.N)__name__ __module__ __qualname____doc__rr/lib/python3.10/site-packages/aiohttp/streams.pyr%src@sBeZdZdegeefddfddZd ddZdefd d ZdS) AsyncStreamIterator read_funcreturnNcC ||_dSN)r )selfr rrr__init__* zAsyncStreamIterator.__init__AsyncStreamIterator[_T]cC|Sr#rr$rrr __aiter__-zAsyncStreamIterator.__aiter__cs8z |IdH}Wn tytw|dkrt|SN)r rStopAsyncIterationr$Zrvrrr __anext__0s zAsyncStreamIterator.__anext__)r!r') rrrrrrr%r*r0rrrrr)s rc@s6eZdZd ddZd ddZdeeeffd d ZdS) ChunkTupleAsyncStreamIteratorstreamrr!NcCr"r#)_stream)r$r2rrrr%;r&z&ChunkTupleAsyncStreamIterator.__init__cCr(r#rr)rrrr*>r+z'ChunkTupleAsyncStreamIterator.__aiter__cs"|jIdH}|dkrt|S)Nr-F)r3 readchunkr.r/rrrr0As z'ChunkTupleAsyncStreamIterator.__anext__)r2rr!N)r!r1) rrrr%r*r bytesboolr0rrrrr1:s  r1c@sTeZdZdeefddZdedeefddZdeefddZde fd d Z d S) AsyncStreamReaderMixinr!cC t|jSr#)rreadliner)rrrr*Ir&z AsyncStreamReaderMixin.__aiter__ncstfddS)z>Returns an asynchronous iterator that yields chunks of size n.cs Sr#)readrr;r$rrNs z5AsyncStreamReaderMixin.iter_chunked..)rr$r;rr=r iter_chunkedLsz#AsyncStreamReaderMixin.iter_chunkedcCr9)z3Yield all available data as soon as it is received.)rreadanyr)rrriter_anyPs zAsyncStreamReaderMixin.iter_anycCst|S)zYield chunks of data as they are received by the server. The yielded objects are tuples of (bytes, bool) as returned by the StreamReader.readchunk method. )r1r)rrr iter_chunksTsz"AsyncStreamReaderMixin.iter_chunksN) rrrrr6r*intr@rBr1rCrrrrr8Hs r8c @seZdZdZdZddddededeedee j d df d d Z d e fd d Z d eeeffddZd eefddZefdeded dfddZdegdfd dfddZdDddZd efddZd efddZdDdd Zd!ed dfd"d#ZdEd!ed$ed dfd%d&ZdDd'd(ZdDd)d*Zd+e d dfd,d-Z d efd.d/Z!dFd1ed efd2d3Z"dGd5ed efd6d7Z#d efd8d9Z$d eeeffd:d;Z%d5ed efdd?Z'd5ed efd@dAZ(d5ed efdBdCZ)dS)Hra*An enhancement of asyncio.StreamReader. Supports asynchronous iteration by line, chunk or as available:: async for line in reader: ... async for chunk in reader.iter_chunked(1024): ... async for slice in reader.iter_any(): ... rN)timerloopprotocollimitrErFr!cCs||_||_|d|_|durt}||_d|_d|_d|_t |_ d|_ d|_ d|_d|_d|_|dur:tn||_g|_dS)NrF) _protocol _low_water _high_waterasyncioZget_event_loop_loop_size_cursor_http_chunk_splits collectionsdeque_buffer_buffer_offset_eof_waiter _eof_waiter _exceptionr_timer_eof_callbacks)r$rGrHrErFrrrr%ms"   zStreamReader.__init__cCs|jjg}|jr|d|j|jr|d|jdkr(|d|j|jf|jr3|d|j|jr>|d|jdd |S) Nz%d byteseofizlow=%d high=%dzw=%rze=%r<%s> ) __class__rrOappendrVrKrLrWrYjoin)r$inforrr__repr__s   zStreamReader.__repr__cCs |j|jfSr#)rKrLr)rrrget_read_buffer_limits z#StreamReader.get_read_buffer_limitscC|jSr#rYr)rrr exceptionzStreamReader.exceptionexc exc_causecCsX||_|j|j}|durd|_t||||j}|dur*d|_t|||dSdSr#)rYr[clearrWrrXr$rjrkwaiterrrrrs  zStreamReader.set_exceptioncallbackcCsB|jrz|WdStytdYdSw|j|dSNException in eof callback)rV Exceptionrrhr[r`r$rorrron_eofs  zStreamReader.on_eofc Csd|_|j}|durd|_t|d|j}|dur!d|_t|d|jD]}z|Wq$ty9tdYq$w|jdS)NTrq) rVrWrrXr[rrrrhrl)r$rncbrrrfeed_eofs      zStreamReader.feed_eofcCrf)z&Return True if 'feed_eof' was called.rVr)rrris_eofszStreamReader.is_eofcC|jo|j S)z=Return True if the buffer is empty and 'feed_eof' was called.rVrTr)rrrat_eofszStreamReader.at_eofcsH|jrdS|jdus J|j|_z |jIdHWd|_dSd|_wr#)rVrXrN create_futurer)rrrwait_eofs zStreamReader.wait_eofdatacCsxtjdtdd|s dS|jr|jd|jd|jd<d|_|jt|7_|jt|8_|j|d|_ dS)zDrollback reading some data from stream, inserting it to buffer head.zJunread_data() is deprecated and will be removed in future releases (#3260)rI) stacklevelNr) warningswarnDeprecationWarningrUrTrOlenrP appendleft _eof_counter)r$r~rrr unread_datas  zStreamReader.unread_datasizecCs|jrJd|s dS|jt|7_|j||jt|7_|j}|dur2d|_t|d|j|jkrC|j j sE|j dSdSdS)Nzfeed_data after feed_eof) rVrOrrTr` total_bytesrWrrLrJ_reading_paused pause_readingr$r~rrnrrr feed_datas  zStreamReader.feed_datacCs&|jdur|jr tdg|_dSdS)Nz?Called begin_http_chunk_receiving whensome data was already fed)rQr RuntimeErrorr)rrrbegin_http_chunk_receivings  z'StreamReader.begin_http_chunk_receivingcCsh|jdur td|jr|jdnd}|j|krdS|j|j|j}|dur2d|_t|ddSdS)NzFCalled end_chunk_receiving without calling begin_chunk_receiving firstr)rQrrr`rWr)r$posrnrrrend_http_chunk_receiving s  z%StreamReader.end_http_chunk_receiving func_namecsz|jdur td||j}|_z$|j |IdHWdn1s(wYWd|_dSWd|_dSd|_w)NzH%s() called while another coroutine is already waiting for incoming data)rWrrNr|rZ)r$rrnrrr_wait*s   zStreamReader._waitcs|IdHSr#) readuntilr)rrrr:<szStreamReader.readline separatorc st|}|dkr td|jdur|jd}d}d}|rm|jr[|r[|j}|jd||d}||r<|||dnd}||7}|t|7}|rMd}||jkrVtd|jr[|s"|jra |S|rk| d IdH|s|S) Nrz,Separator should be at least one-byte stringr-Tr rFz Chunk too bigr) r ValueErrorrYrTrUfind_read_nowait_chunkrLrVr) r$rZseplenchunkZ chunk_sizeZ not_enoughoffsetZicharr~rrrr?s<     zStreamReader.readuntilrr;cs|jdur |j |jr%|js%t|ddd|_|jdkr%tjddd|s)dS|dkrEg} |IdH}|s:n||q0d |S|jsY|jsY| d IdH|jsY|jrK| |S) NTrrr zEMultiple access to StreamReader in eof state, might be infinite loop.)Z stack_infor-r<) rYrVrTgetattrrrwarningrAr`rar _read_nowait)r$r;blocksblockrrrr<cs4       zStreamReader.readcsD|jdur |j|js|js|dIdH|js|jr|dS)NrAr)rYrTrVrrr)rrrrAs   zStreamReader.readanycs |jdur |j|jr1|jd}||jkrdS||jkr)|||jdfStd|js |jr;|ddfS|j r@dS| d IdHq) aReturns a tuple of (data, end_of_http_chunk). When chunked transfer encoding is used, end_of_http_chunk is a boolean indicating if the end of the data corresponds to the end of a HTTP chunk , otherwise it is always False. TNrr-TzESkipping HTTP chunk end due to data consumption beyond chunk boundaryrFr4r5) rYrQpoprPrrrrTrrVr)r$rrrrr5s(     zStreamReader.readchunkcsx|jdur |jg}|dkr7||IdH}|s(d|}t|t|||||t|8}|dksd|S)Nrr-)rYr<rarMIncompleteReadErrorrr`)r$r;rrpartialrrr readexactlys     zStreamReader.readexactlycCs2|jdur|j|jr|jstd||S)Nz9Called while some coroutine is waiting for incoming data.)rYrWZdonerrr?rrr read_nowaits  zStreamReader.read_nowaitcCs|jd}|j}|dkr$t|||kr$||||}|j|7_n|r5|j||d}d|_n|j}|jt|8_|jt|7_|j}|rf|d|jkrf|d|rf|d|jksX|j|jkru|j j ru|j |S)Nrr) rTrUrpopleftrOrPrQrrKrJrresume_reading)r$r;Z first_bufferrr~Z chunk_splitsrrrrs&      zStreamReader._read_nowait_chunkcCs^|jg}|jr&||}|||dkr#|t|8}|dkr#n|js |r-d|SdS)z6Read not more than n bytes, or whole buffer if n == -1rrr-)rZZassert_timeoutrTrr`rra)r$r;chunksrrrrrs    zStreamReader._read_nowaitr!Nr)rr)*rrrrrr rDrrrMAbstractEventLoopr%strrcr rd BaseExceptionrhr rrrtrvr7rxr{r}r6rrrrrr:rr<rAr5rrrrrrrrr]s\        $* "rc@seZdZd*ddZdefddZdeefddZe fd ed eddfd d Z d e gdfddfddZ d*ddZ defddZdefddZd*ddZd+dededdfddZdefddZd,dedefd d!Zdefd"d#Zdeeeffd$d%Zdedefd&d'Zd,dedefd(d)ZdS)-EmptyStreamReaderr!NcCs d|_dS)NFZ_read_eof_chunkr)rrrr%r&zEmptyStreamReader.__init__cCs d|jjS)Nr])r_rr)rrrrcrezEmptyStreamReader.__repr__cCdSr#rr)rrrrh r+zEmptyStreamReader.exceptionrjrkcCrr#r)r$rjrkrrrrszEmptyStreamReader.set_exceptionrocCs,z|WdStytdYdSwrp)rrrrhrsrrrrts   zEmptyStreamReader.on_eofcCrr#rr)rrrrvr+zEmptyStreamReader.feed_eofcCdSNTrr)rrrrxr+zEmptyStreamReader.is_eofcCrrrr)rrrr{!r+zEmptyStreamReader.at_eofcsdSr#rr)rrrr}$zEmptyStreamReader.wait_eofrr~r;cCrr#r)r$r~r;rrrr'r+zEmptyStreamReader.feed_datacdSr,rr)rrrr:*rzEmptyStreamReader.readlinercrr,rr?rrrr<-rzEmptyStreamReader.readcrr,rr)rrrrA2rzEmptyStreamReader.readanycs|js d|_dSdS)NTr4rrr)rrrr55s zEmptyStreamReader.readchunkcstd|r,)rMrr?rrrr<s zEmptyStreamReader.readexactlycCrr,rr?rrrr?r+zEmptyStreamReader.read_nowaitrrr)rrrr%rrcrrrhr rrrtrvr7rxr{r}r6rDrr:r<rAr r5rrrrrrrs0    rrc@seZdZdZdejddfddZdefddZde fd d Z de fd d Z de e fd dZefde de ddfddZddededdfddZdddZdefddZdeefddZdS) rz>DataQueue is a general-purpose blocking queue with one reader.rFr!NcCs,||_d|_d|_d|_d|_t|_dS)NFr)rNrVrWrYrOrRrSrT)r$rFrrrr%Is zDataQueue.__init__cCr9r#)rrTr)rrr__len__Qr&zDataQueue.__len__cCrfr#rwr)rrrrxTrizDataQueue.is_eofcCryr#rzr)rrrr{WszDataQueue.at_eofcCrfr#rgr)rrrrhZrizDataQueue.exceptionrjrkcCs4d|_||_|j}|durd|_t|||dSdSr)rVrYrWrrmrrrr]szDataQueue.set_exceptionrr~rcCsD|j|7_|j||f|j}|dur d|_t|ddSdSr#)rOrTr`rWrrrrrrjszDataQueue.feed_datacCs,d|_|j}|durd|_t|ddSdSr)rVrWr)r$rnrrrrvss zDataQueue.feed_eofc s|js*|js*|jr J|j|_z|jIdHWntjtjfy)d|_w|jr=|j\}}|j |8_ |S|j durE|j t r#) rTrVrWrNr|rMZCancelledError TimeoutErrorrrOrYrr$r~rrrrr<{s"    zDataQueue.readcCr9r#)rr<r)rrrr*r&zDataQueue.__aiter__rr)rrrrrMrr%rDrr7rxr{rrrhr rrrrvr<rr*rrrrrFs&  rcsbeZdZdZdededejddffdd Zdd e d eddffd d Z de ffdd Z Z S)rzlFlowControlDataQueue resumes and pauses an underlying stream. It is a destination for parsed data. rGrHrFr!Ncs"tj|d||_|d|_dS)N)rFrI)superr%rJ_limit)r$rGrHrFr_rrr%szFlowControlDataQueue.__init__rr~rcs8t|||j|jkr|jjs|jdSdSdSr#)rrrOrrJrrrrrrrszFlowControlDataQueue.feed_datacs\ztIdHW|j|jkr|jjr|jSSS|j|jkr,|jjr-|jwwwr#)rr<rOrrJrrr)rrrr<s  zFlowControlDataQueue.readr) rrrrr rDrMrr%rrr< __classcell__rrrrrsr)$rMrRrtypingrrrrrrrr r Z base_protocolr Zhelpersr rrrrlogr__all__rrrrrr1r8rrr__annotations__rrrrrrs* ,  *?M