U g~&@sddlmZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z ddlmZmZmZmZddlmZdgZed Ze d ZGd d d e eZGd ddZdS)) annotationsN)Any AsyncIteratorCallableGenericIterableTypeVar)ConcurrencyError) OP_BINARYOP_CONTOP_TEXTFrame)Data Assemblerzutf-8Tc@sheZdZdZddddZddddZd dd d d Zd dd dZdddddZddddZ dS) SimpleQueuez Simplified version of :class:`asyncio.Queue`. Provides only the subset of functionality needed by :class:`Assembler`. NonereturncCst|_d|_t|_dSN)asyncioZget_running_looploop get_waiter collectionsdequequeueselfr?/tmp/pip-unpacked-wheel-dx_q7dq3/websockets/asyncio/messages.py__init__#s zSimpleQueue.__init__intcCs t|jSr)lenrrrrr __len__(szSimpleQueue.__len__r)itemrcCs0|j||jdk r,|js,|jddS)z+Put an item into the queue without waiting.N)rappendrdoneZ set_result)rr%rrr put+s zSimpleQueue.putcsR|jsH|jdk rtd|j|_z|jIdHW5|jd|_X|jS)z?Remove and return an item from the queue, waiting if necessary.Nzget is already running)rrr rZ create_futurecancelpopleftrrrr get1s   zSimpleQueue.getz Iterable[T])itemsrcCs0|jdkstd|jr td|j|dS)z)Put back items into an empty, idle queue.Nz%cannot reset() while get() is runningz&cannot reset() while queue isn't empty)rAssertionErrorrextend)rr,rrr reset>szSimpleQueue.resetcCs2|jdk r$|js$|jtd|jdS)Nstream of frames ended)rr'Z set_exceptionEOFErrorrclearrrrr abortDszSimpleQueue.abortN) __name__ __module__ __qualname____doc__r!r$r(r+r/r3rrrr rs rc@seZdZdZddddddfddd d d d d d Zd!dddddZd"dddddZdd dddZd dddZd dddZ d ddd Z dS)#ra Assemble messages from frames. :class:`Assembler` expects only data frames. The stream of frames must respect the protocol; if it doesn't, the behavior is undefined. Args: pause: Called when the buffer of frames goes above the high water mark; should pause reading from the network. resume: Called when the buffer of frames goes below the low water mark; should resume reading from the network. NcCsdSrrrrrr _zAssembler.cCsdSrrrrrr r9`r:r"z int | NonezCallable[[], Any]r)highlowpauseresumercCsht|_|dkr|d}|dkr(td||kr8td|||_|_||_||_d|_d|_d|_ dS)Nrz%low must be positive or equal to zeroz)high must be greater than or equal to lowF) rframes ValueErrorr;r<r=r>pausedget_in_progressclosed)rr;r<r=r>rrr r![szAssembler.__init__z bool | Noner)decodercs&|jrtd|jrtdd|_z|jIdH}Wntjk rTd|_YnX||j t ksv|j t ksvt |dkr|j t k}|g}|j sz|jIdH}Wn*tjk r|j|d|_YnX||j tkst ||qd|_ddd|D}|r|S|SdS) a Read the next message. :meth:`get` returns a single :class:`str` or :class:`bytes`. If the message is fragmented, :meth:`get` waits until the last frame is received, then it reassembles the message and returns it. To receive messages frame by frame, use :meth:`get_iter` instead. Args: decode: :obj:`False` disables UTF-8 decoding of text frames and returns :class:`bytes`. :obj:`True` forces UTF-8 decoding of binary frames and returns :class:`str`. Raises: EOFError: If the stream of frames has ended. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. r0&get() or get_iter() is already runningTNFr:css|] }|jVqdSr)data).0framerrr sz Assembler.get..)rDr1rCr r@r+rCancelledError maybe_resumeopcoder r r-finr/r r&joinrE)rrErIr@rGrrr r+zs<   z Assembler.getzAsyncIterator[Data]cCs|jrtd|jrtdd|_z|jIdH}Wntjk rTd|_YnX||j t ksv|j t ksvt |dkr|j t k}|rt }||j|jVn|jV|js|jIdH}||j tkst |r||j|jVq|jVqd|_dS)ap Stream the next message. Iterating the return value of :meth:`get_iter` asynchronously yields a :class:`str` or :class:`bytes` for each frame in the message. The iterator must be fully consumed before calling :meth:`get_iter` or :meth:`get` again. Else, :exc:`ConcurrencyError` is raised. This method only makes sense for fragmented messages. If messages aren't fragmented, use :meth:`get` instead. Args: decode: :obj:`False` disables UTF-8 decoding of text frames and returns :class:`bytes`. :obj:`True` forces UTF-8 decoding of binary frames and returns :class:`str`. Raises: EOFError: If the stream of frames has ended. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. r0rFTNF)rDr1rCr r@r+rrKrLrMr r r- UTF8DecoderrErGrNr )rrErIdecoderrrr get_iters4  zAssembler.get_iterr)rIrcCs&|jrtd|j||dS)z Add ``frame`` to the next message. Raises: EOFError: If the stream of frames has ended. r0N)rDr1r@r( maybe_pause)rrIrrr r(s z Assembler.putrcCs(t|j|jkr$|js$d|_|dS)z7Pause the writer if queue is above the high water mark.TN)r#r@r;rBr=rrrr rS szAssembler.maybe_pausecCs(t|j|jkr$|jr$d|_|dS)z7Resume the writer if queue is below the low water mark.FN)r#r@r<rBr>rrrr rLszAssembler.maybe_resumecCs|jr dSd|_|jdS)z End the stream of frames. Callling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`, or :meth:`put` is safe. They will raise :exc:`EOFError`. NT)rDr@r3rrrr closeszAssembler.close)N)N) r4r5r6r7r!r+rRr(rSrLrTrrrr rKs@A) __future__rrcodecsrtypingrrrrrr exceptionsr r@r r r rr__all__getincrementaldecoderrPrrrrrrr s    0