o RhR@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__rrQ/home/air/sanwanet/backup_V2/venv/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$rvrrr __anext__0s zAsyncStreamIterator.__anext__)r!r') rrrrrrr%r*r1rrrrr)s rc@s6eZdZd ddZd ddZdeeeffd d ZdS) ChunkTupleAsyncStreamIteratorstreamrr!NcCr"r#)_stream)r$r3rrrr%;r&z&ChunkTupleAsyncStreamIterator.__init__cCr(r#rr)rrrr*>r+z'ChunkTupleAsyncStreamIterator.__aiter__cs"|jIdH}|dkrt|S)Nr-F)r4 readchunkr.r/rrrr1As z'ChunkTupleAsyncStreamIterator.__anext__)r3rr!N)r!r2) rrrr%r*r bytesboolr1rrrrr2:s  r2c@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_chunkedcCr:)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. )r2r)rrr iter_chunksTsz"AsyncStreamReaderMixin.iter_chunksN) rrrrr7r*intrArCr2rDrrrrr9Hs r9c @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)timerloopprotocollimitrFrGr!cCs||_||_|d|_|durt}||_d|_d|_d|_t |_ d|_ d|_ d|_d|_d|_|dur:tn||_g|_dS)NrF) _protocol _low_water _high_waterasyncioget_event_loop_loop_size_cursor_http_chunk_splits collectionsdeque_buffer_buffer_offset_eof_waiter _eof_waiter _exceptionr_timer_eof_callbacks)r$rHrIrFrGrrrr%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__rrQappendrXrLrMrYr[join)r$inforrr__repr__s   zStreamReader.__repr__cCs |j|jfSr#)rLrMr)rrrget_read_buffer_limits z#StreamReader.get_read_buffer_limitscC|jSr#r[r)rrr exceptionzStreamReader.exceptionexc exc_causecCsX||_|j|j}|durd|_t||||j}|dur*d|_t|||dSdSr#)r[r]clearrYrrZr$rlrmwaiterrrrrs  zStreamReader.set_exceptioncallbackcCsB|jrz|WdStytdYdSw|j|dSNException in eof callback)rX Exceptionrrjr]rbr$rqrrron_eofs  zStreamReader.on_eofc Csd|_|j}|durd|_t|d|j}|dur!d|_t|d|jD]}z|Wq$ty9tdYq$w|jdS)NTrs) rXrYrrZr]rtrrjrn)r$rpcbrrrfeed_eofs      zStreamReader.feed_eofcCrh)z&Return True if 'feed_eof' was called.rXr)rrris_eofszStreamReader.is_eofcC|jo|j S)z=Return True if the buffer is empty and 'feed_eof' was called.rXrVr)rrrat_eofszStreamReader.at_eofcsH|jrdS|jdus J|j|_z |jIdHWd|_dSd|_wr#)rXrZrP 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)rJ) stacklevelNr) warningswarnDeprecationWarningrWrVrQlenrR appendleft _eof_counter)r$rrrr 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) rXrQrrVrb total_bytesrYrrMrK_reading_paused pause_readingr$rrrprrr feed_datas  zStreamReader.feed_datacCs&|jdur|jr tdg|_dSdS)Nz?Called begin_http_chunk_receiving whensome data was already fed)rSr 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)rSrrrbrYr)r$posrprrrend_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)rYrrPr~r\)r$rrprrr_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 ValueErrorr[rVrWfind_read_nowait_chunkrMrXr) r$rseplenchunk chunk_size not_enoughoffseticharrrrrr?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.) stack_infor-r=) r[rXrVgetattrrrwarningrBrbrcr _read_nowait)r$r<blocksblockrrrr=cs4       zStreamReader.readcsD|jdur |j|js|js|dIdH|js|jr|dS)NrBr)r[rVrXrrr)rrrrBs   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 boundaryrFr5r6) r[rSpoprRrrrrVrrXr)r$rrrrr6s(     zStreamReader.readchunkcsx|jdur |jg}|dkr7||IdH}|s(d|}t|t|||||t|8}|dksd|S)Nrr-)r[r=rcrNIncompleteReadErrorrrb)r$r<rrpartialrrr readexactlys     zStreamReader.readexactlycCs2|jdur|j|jr|jstd||S)Nz9Called while some coroutine is waiting for incoming data.)r[rYdonerrr@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) rVrWrpopleftrQrRrSrrLrKrresume_reading)r$r< first_bufferrr 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-)r\assert_timeoutrVrrbrrc)r$r<chunksrrrrrs    zStreamReader._read_nowaitr!Nr)rr)*rrrrrr rErrrNAbstractEventLoopr%strrer rf BaseExceptionrjr rrrvrxr8rzr}rr7rrrrrr;rr=rBr6rrrrrrrrr]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)NF_read_eof_chunkr)rrrr%r&zEmptyStreamReader.__init__cCs d|jjS)Nr_)rarr)rrrrergzEmptyStreamReader.__repr__cCdSr#rr)rrrrj r+zEmptyStreamReader.exceptionrlrmcCrr#r)r$rlrmrrrrszEmptyStreamReader.set_exceptionrqcCs,z|WdStytdYdSwrr)rtrrjrurrrrvs   zEmptyStreamReader.on_eofcCrr#rr)rrrrxr+zEmptyStreamReader.feed_eofcCdSNTrr)rrrrzr+zEmptyStreamReader.is_eofcCrrrr)rrrr}!r+zEmptyStreamReader.at_eofcsdSr#rr)rrrr$zEmptyStreamReader.wait_eofrrr<cCrr#r)r$rr<rrrr'r+zEmptyStreamReader.feed_datacdSr,rr)rrrr;*rzEmptyStreamReader.readlinercrr,rr@rrrr=-rzEmptyStreamReader.readcrr,rr)rrrrB2rzEmptyStreamReader.readanycs|js d|_dSdS)NTr5rrr)rrrr65s zEmptyStreamReader.readchunkcstd|r,)rNrr@rrrr<s zEmptyStreamReader.readexactlycCrr,rr@rrrr?r+zEmptyStreamReader.read_nowaitrrr)rrrr%rrerrrjr rrrvrxr8rzr}rr7rErr;r=rBr r6rrrrrrrs0    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.rGr!NcCs,||_d|_d|_d|_d|_t|_dS)NFr)rPrXrYr[rQrTrUrV)r$rGrrrr%Is zDataQueue.__init__cCr:r#)rrVr)rrr__len__Qr&zDataQueue.__len__cCrhr#ryr)rrrrzTrkzDataQueue.is_eofcCr{r#r|r)rrrr}WszDataQueue.at_eofcCrhr#rir)rrrrjZrkzDataQueue.exceptionrlrmcCs4d|_||_|j}|durd|_t|||dSdSr)rXr[rYrrorrrr]szDataQueue.set_exceptionrrrcCsD|j|7_|j||f|j}|dur d|_t|ddSdSr#)rQrVrbrYrrrrrrjszDataQueue.feed_datacCs,d|_|j}|durd|_t|ddSdSr)rXrYr)r$rprrrrxss zDataQueue.feed_eofc s|js*|js*|jr J|j|_z|jIdHWntjtjfy)d|_w|jr=|j\}}|j |8_ |S|j durE|j t r#) rVrXrYrPr~rNCancelledError TimeoutErrorrrQr[rr$rrrrrr={s"    zDataQueue.readcCr:r#)rr=r)rrrr*r&zDataQueue.__aiter__rr)rrrrrNrr%rErr8rzr}rrrjr rrrrxr=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. rHrIrGr!Ncs"tj|d||_|d|_dS)N)rGrJ)superr%rK_limit)r$rHrIrGrarrr%szFlowControlDataQueue.__init__rrrcs8t|||j|jkr|jjs|jdSdSdSr#)rrrQrrKrrrrrrrszFlowControlDataQueue.feed_datacs\ztIdHW|j|jkr|jjr|jSSS|j|jkr,|jjr-|jwwwr#)rr=rQrrKrrr)rrrr=s  zFlowControlDataQueue.readr) rrrrr rErNrr%rrr= __classcell__rrrrrsr)$rNrTrtypingrrrrrrrr r base_protocolr helpersr rrrrlogr__all__rrtrrr2r9rrr__annotations__rrrrrrs* ,  *?M