Z?A @ s d d l Z d d l Z d d l m Z m Z d d l m Z d d d d d f Z d
d Z Gd d d e Z
Gd
d d Z Gd d d e Z Gd d d Z
Gd d d e
Z Gd d d e
Z e Z Gd d d Z Gd d d e Z d S) N )
set_exception
set_result)internal_logger
EMPTY_PAYLOAD EofStream StreamReader DataQueueFlowControlDataQueue c @ s e Z d Z d Z d S)r zeof stream indication.N)__name__
__module__ __qualname____doc__ r r /streams.pyr s c @ s4 e Z d Z d d Z d d Z d d Z d S)AsyncStreamIteratorc C s
| | _ d S)N) read_func)selfr r r r __init__ s z AsyncStreamIterator.__init__c C s | S)Nr )r r r r __aiter__ s z AsyncStreamIterator.__aiter__c sF y | j Id H} Wn t k
r/ t Yn X| d k rB t | S)N )r r StopAsyncIteration)r rvr r r __anext__ s
z AsyncStreamIterator.__anext__N)r
r r r r r r r r r r s r c @ s e Z d Z d d Z d S) ChunkTupleAsyncStreamIteratorc s' | j Id H} | d k r# t | S)Nr F)r F)r r )r r r r r r & s z'ChunkTupleAsyncStreamIterator.__anext__N)r
r r r r r r r r % s r c @ s@ e Z d Z d d Z d d Z d d Z d d Z d S)
AsyncStreamReaderMixinc C s
t | j S)N)r readline)r r r r r / s z AsyncStreamReaderMixin.__aiter__c s t f d d S)zzReturns an asynchronous iterator that yields chunks of size n.
Python-3.5 available for Python 3.5+ only
c s
j S)N)readr )nr r r <lambda>7 s z5AsyncStreamReaderMixin.iter_chunked.<locals>.<lambda>)r )r r r )r r r iter_chunked2 s z#AsyncStreamReaderMixin.iter_chunkedc C s
t | j S)zReturns an asynchronous iterator that yields all the available
data as soon as it is received
Python-3.5 available for Python 3.5+ only
)r readany)r r r r iter_any9 s z AsyncStreamReaderMixin.iter_anyc C s
t | j S)a
Returns an asynchronous iterator that yields 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.
Python-3.5 available for Python 3.5+ only
)r readchunk)r r r r iter_chunksA s z"AsyncStreamReaderMixin.iter_chunksN)r
r r r r" r$ r& r r r r r - s r c @ s? e Z d Z d Z d Z d e d d d d d d Z d d
Z d d Z d
d Z d d Z
d d Z d d Z d d Z
d d Z d d Z d d d Z d d Z d d Z d! d" Z d# d$ Z d4 d& d' Z d( d) Z d* d+ Z d, d- Z d5 d. d/ Z d0 d1 Z d2 d3 Z d S)6r a* 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():
...
r limittimerNloopc C s | | _ | | _ | d | _ | d k r7 t j } | | _ d | _ d | _ d | _ t j
| _ d | _ d | _
d | _ d | _ d | _ | | _ g | _ d S)Nr r F) _protocol
_low_water _high_waterasyncioZget_event_loop_loop_size_cursor_http_chunk_splits collectionsdeque_buffer_buffer_offset_eof_waiter _eof_waiter
_exception_timer_eof_callbacks)r protocolr' r( r) r r r r [ s"
zStreamReader.__init__c C s | j j g } | j r, | j d | j | j rB | j d | j t k rn | j d | j | j f | j r | j d | j | j r | j d | j d d j
| S)Nz%d byteseofzlow=%d high=%dzw=%rze=%rz<%s> ) __class__r
r/ appendr6 r+
DEFAULT_LIMITr, r7 r9 join)r infor r r __repr__o s
zStreamReader.__repr__c C s | j S)N)r9 )r r r r exception} s zStreamReader.exceptionc C sp | | _ | j j | j } | d k rA d | _ t | | | j } | d k rl t | | d | _ d S)N)r9 r; clearr7 r r8 )r excwaiterr r r r s
zStreamReader.set_exceptionc C sM | j r9 y | WqI t k
r5 t j d YqI Xn | j j | d S)NzException in eof callback)r6 Exceptionr rE r; r@ )r callbackr r r on_eof s
zStreamReader.on_eofc C s d | _ | j } | d k r4 d | _ t | d | j } | d k r_ d | _ t | d x> | j D]3 } y | Wqi t k
r t j d Yqi Xqi W| j j d S)NTzException in eof callback) r6 r7 r r8 r; rI r rE rF )r rH cbr r r feed_eof s
zStreamReader.feed_eofc C s | j S)z&Return True if 'feed_eof' was called.)r6 )r r r r is_eof s zStreamReader.is_eofc C s | j o | j S)z=Return True if the buffer is empty and 'feed_eof' was called.)r6 r4 )r r r r at_eof s zStreamReader.at_eofc s@ | j r
d S| j j | _ z | j Id HWd d | _ Xd S)N)r6 r.
create_futurer8 )r r r r wait_eof s zStreamReader.wait_eofc C s | s
d S| j r= | j d | j d | j d <d | _ | j t | 7_ | j t | 8_ | j j | d | _ d S)zN rollback reading some data from stream, inserting it to buffer head.
Nr )r5 r4 r/ lenr0
appendleftZ _eof_counter)r datar r r unread_data s ! zStreamReader.unread_datac C s | s
d S| j t | 7_ | j j | | j t | 7_ | j } | d k ro d | _ t | d | j | j k r | j j r | j j
d S)NF) r/ rR r4 r@ total_bytesr7 r r, r* _reading_paused
pause_reading)r rT sizerH r r r feed_data s
zStreamReader.feed_datac C s | j d k r g | _ d S)N)r1 )r r r r begin_http_chunk_receiving s z'StreamReader.begin_http_chunk_receivingc C sR | j d k r t d | j s; | j d | j k rN | j j | j d S)NzFCalled end_chunk_receiving without calling begin_chunk_receiving firstr )r1 RuntimeErrorrV r@ )r r r r end_http_chunk_receiving s
z%StreamReader.end_http_chunk_receivingc sy | j d k r t d | | j j } | _ z3 | j r^ | j | Id HWd QRXn | Id HWd d | _ Xd S)NzH%s() called while another coroutine is already waiting for incoming data)r7 r] r. rP r: )r func_namerH r r r _wait s
zStreamReader._waitc s | j d k r | j g } d } d } x | r x | j r | r | j } | j d j d | d } | j | r | | n d } | j | | t | 7} | r d } | | j k r6 t d q6 W| j r P| r- | j
d Id Hq- Wd j | S)
Nr Ts
r FzLine is too longr r r\ ) r9 r4 r5 find_read_nowait_chunkr@ rR r,
ValueErrorr6 r` rB )r lineZ line_sizeZ
not_enoughoffsetZicharrT r r r r s*
zStreamReader.readliner c s | j d k r | j | s" d S| d k rm g } x) | j Id H} | sO P| j | q7 Wd j | S| j r | j r | j d Id H| j | S)Nr r r )r9 r# r@ rB r4 r6 r` _read_nowait)r r blocksblockr r r r s
zStreamReader.readc sK | j d k r | j | j r> | j r> | j d Id H| j d S)Nr# r r\ )r9 r4 r6 r` rf )r r r r r# C s
zStreamReader.readanyc s | j d k r | j | j rw | j rw | j re | j | j d k re | j d d | _ d S| j d Id H| j s d S| j d k rxY | j r | j d } | j d d | _ | | j k r | j | | j d f Sq W| j d
d f S| j d d f Sd S) zReturns 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.
Nr r r Tr% F)r T)r Fr\ r\ )r9 r4 r6 r1 r0 r` rf rb )r posr r r r% L s$
zStreamReader.readchunkc s | j d k r | j g } xu | d k r | j | Id H} | su d j | } t j j | t | | | j | | t | 8} q! Wd j | S)Nr r )r9 r rB r- streamsIncompleteReadErrorrR r@ )r r rg rh partialr r r readexactlyj s
zStreamReader.readexactlyc C sJ | j d k r | j | j r= | j j r= t d | j | S)Nz9Called while some coroutine is waiting for incoming data.)r9 r7 doner] rf )r r r r r read_nowaitz s zStreamReader.read_nowaitc C s | j d } | j } | d k r^ t | | | k r^ | | | | } | j | 7_ n> | r | j j | | d } d | _ n | j j } | j t | 8_ | j t | 7_ | j | j k r | j j r | j j | S)Nr r r\ )
r4 r5 rR popleftr/ r0 r+ r* rW resume_reading)r r Z first_bufferre rT r r r rb s
"
z StreamReader._read_nowait_chunkc C sr g } xR | j rZ | j | } | j | | d k r | t | 8} | d k r Pq W| rn d j | Sd S)Nr r r r\ )r4 rb r@ rR rB )r r Zchunkschunkr r r rf s
zStreamReader._read_nowaitr\ r\ ) r
r r r rV rA r rD rE r rK rM rN rO rQ rU rZ r[ r^ r` r r r# r% rm ro rb rf r r r r r K s2 % c @ s e Z d Z d d Z d d Z d d Z d d Z d d
Z d d Z d
d Z d d Z
d d Z d d d Z d d Z
d d Z d d Z d d Z d S) EmptyStreamReaderc C s d S)Nr )r r r r rE s zEmptyStreamReader.exceptionc C s d S)Nr )r rG r r r r s z EmptyStreamReader.set_exceptionc C s1 y | Wn t k
r, t j d Yn Xd S)NzException in eof callback)rI r rE )r rJ r r r rK s
zEmptyStreamReader.on_eofc C s d S)Nr )r r r r rM s zEmptyStreamReader.feed_eofc C s d S)NTr )r r r r rN s zEmptyStreamReader.is_eofc C s d S)NTr )r r r r rO s zEmptyStreamReader.at_eofc s d S)Nr )r r r r rQ s zEmptyStreamReader.wait_eofc C s d S)Nr )r rT r r r rZ s zEmptyStreamReader.feed_datac s d S)Nr r )r r r r r s zEmptyStreamReader.readliner c s d S)Nr r )r r r r r r s zEmptyStreamReader.readc s d S)Nr r )r r r r r# s zEmptyStreamReader.readanyc s d S)Nr F)r Fr )r r r r r% s zEmptyStreamReader.readchunkc s t j j d | d S)Nr )r- rj rk )r r r r r rm s z EmptyStreamReader.readexactlyc C s d S)Nr r )r r r r ro s z EmptyStreamReader.read_nowaitNr\ )r
r r rE r rK rM rN rO rQ rZ r r r# r% rm ro r r r r rs s rs c @ s e Z d Z d Z d d d d Z d d Z d d Z d
d Z d d
Z d d Z d d d Z
d d Z d d Z d d Z
d S)r z>DataQueue is a general-purpose blocking queue with one reader.r) Nc C s@ | | _ d | _ d | _ d | _ d | _ t j | _ d S)NFr )r. r6 r7 r9 r/ r2 r3 r4 )r r) r r r r s zDataQueue.__init__c C s
t | j S)N)rR r4 )r r r r __len__ s zDataQueue.__len__c C s | j S)N)r6 )r r r r rN s zDataQueue.is_eofc C s | j o | j S)N)r6 r4 )r r r r rO s zDataQueue.at_eofc C s | j S)N)r9 )r r r r rE s zDataQueue.exceptionc C sA d | _ | | _ | j } | d k r= t | | d | _ d S)NT)r6 r9 r7 r )r rG rH r r r r s
zDataQueue.set_exceptionr c C sT | j | 7_ | j j | | f | j } | d k rP d | _ t | d d S)NT)r/ r4 r@ r7 r )r rT rY rH r r r rZ s zDataQueue.feed_datac C s8 d | _ | j } | d k r4 d | _ t | d d S)NTF)r6 r7 r )r rH r r r rM s
zDataQueue.feed_eofc s | j rc | j rc | j j | _ y | j Id HWn* t j t j f k
rb d | _ Yn X| j r | j j \ } } | j | 8_ | S| j
d k r | j
n t d S)N) r4 r6 r. rP r7 r- CancelledError TimeoutErrorrp r/ r9 r )r rT rY r r r r s zDataQueue.readc C s
t | j S)N)r r )r r r r r $ s zDataQueue.__aiter__)r
r r r r rt rN rO rE r rZ rM r r r r r r r s c sX e Z d Z d Z d e d d f d d Z f d d Z f d d
Z S) r
zgFlowControlDataQueue resumes and pauses an underlying stream.
It is a destination for parsed data.r' r) Nc s- t j d | | | _ | d | _ d S)Nr) r )superr r* _limit)r r< r' r) )r? r r r - s z FlowControlDataQueue.__init__c sC t j | | | j | j k r? | j j r? | j j d S)N)rw rZ r/ rx r* rW rX )r rT rY )r? r r rZ 3 s z FlowControlDataQueue.feed_datac sI z t j Id HSWd | j | j k rD | j j rD | j j Xd S)N)rw r r/ rx r* rW rq )r )r? r r r 9 s zFlowControlDataQueue.read)r
r r r rA r rZ r r r )r? r r
( s )r- r2 Zhelpersr r logr __all__rA rI r r r r r rs r r r
r r r r <module> s
b0 I