o Yiy+@sddlmZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z m Z mZddlmZddlmZmZmZmZddlmZd gZed Ze d ZGd d d e eZGdd d ZdS)) annotationsN) AsyncIteratorIterable)AnyCallableGenericLiteralTypeVaroverload)ConcurrencyError) OP_BINARYOP_CONTOP_TEXTFrame)Data Assemblerzutf-8Tc@sNeZdZdZdddZdddZdd d ZddddZdddZdddZ dS) SimpleQueuez Simplified version of :class:`asyncio.Queue`. Provides only the subset of functionality needed by :class:`Assembler`. returnNonecCst|_d|_t|_dSN)asyncioZget_running_looploop get_waiter collectionsdequequeueselfr U/home/kim/smarthome/.venv/lib/python3.10/site-packages/websockets/asyncio/messages.py__init__s zSimpleQueue.__init__intcCs t|jSr)lenrrr r r!__len__"s zSimpleQueue.__len__itemrcCs8|j||jdur|js|jddSdSdS)zPut an item into the queue.N)rappendrdoneZ set_result)rr&r r r!put%s zSimpleQueue.putTblockboolcsp|js3|s td|jdusJd|j|_z|jIdHW|jd|_n |jd|_w|jS)z?Remove and return an item from the queue, waiting if necessary.stream of frames endedNzcannot call get() concurrently)rEOFErrorrrZ create_futurecancelpopleft)rr*r r r!get+s    zSimpleQueue.getitems Iterable[T]cCs0|jdus Jd|jrJd|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)rrextend)rr1r r r!reset9szSimpleQueue.resetcCs0|jdur|js|jtddSdSdS)z8Close the queue, raising EOFError in get() if necessary.Nr,)rr(Z set_exceptionr-rr r r!abort?szSimpleQueue.abortNrr)rr#)r&rrr)T)r*r+rr)r1r2rr) __name__ __module__ __qualname____doc__r"r%r)r0r4r5r r r r!rs     rc@seZdZdZddddddfd.ddZed/ddZed0ddZed1d2ddZd1d2ddZed3ddZed4d dZed1d5d"dZd1d5d#dZd6d&d'Zd7d(d)Z d7d*d+Z d7d,d-Z dS)8ra 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. NcCdSrr r r r r!XzAssembler.cCr;rr r r r r!r<Yr=high int | NonelowpauseCallable[[], Any]resumerrcCst|_|dur|dur|d}|dur|dur|d}|dur4|dur4|dkr,td||kr4td|||_|_||_||_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@rArCpausedget_in_progressclosed)rr>r@rArCr r r!r"Ts  zAssembler.__init__decode Literal[True]strcdSrr rrJr r r!r0vz Assembler.getLiteral[False]bytescrMrr rNr r r!r0yrO bool | NonercrMrr rNr r r!r0|rOcs|jrtdd|_z_|j|j IdH}||jtus'|jtus'J|dur0|jtu}|g}|j sfz |j|j IdH}Wnt j yR|j |w||jt us^J|||j r6Wd|_nd|_wddd|D}|r|S|S)a0 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. UnicodeDecodeError: If a text frame contains invalid UTF-8. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. &get() or get_iter() is already runningTNFcss|]}|jVqdSr)data).0framer r r! sz Assembler.get..)rHr rEr0rI maybe_resumeopcoderr finrCancelledErrorr4rr'joinrJ)rrJrWrErUr r r!r0s8    AsyncIterator[str]cCr;rr rNr r r!get_iterzAssembler.get_iterAsyncIterator[bytes]cCr;rr rNr r r!r_r`AsyncIterator[Data]cCr;rr rNr r r!r_r`cCs |jrtdd|_z |j|j IdH}Wn tjy$d|_w||jt us5|jt us5J|dur>|jt u}|rMt }| |j |jVnt|j V|js|j|j IdH}||jtuslJ|rx| |j |jVnt|j V|jrVd|_dS)a 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. UnicodeDecodeError: If a text frame contains invalid UTF-8. ConcurrencyError: If two coroutines run :meth:`get` or :meth:`get_iter` concurrently. rSTNF)rHr rEr0rIrr\rYrZrr UTF8DecoderrJrUr[rQr)rrJrWdecoderr r r!r_s6     rWrcCs&|jrtd|j||dS)z Add ``frame`` to the next message. Raises: EOFError: If the stream of frames has ended. r,N)rIr-rEr) maybe_pause)rrWr r r!r) s  z Assembler.putcCs>|jdurdSt|j|jkr|jsd|_|dSdSdS)z7Pause the writer if queue is above the high water mark.NT)r>r$rErGrArr r r!re  zAssembler.maybe_pausecCs>|jdurdSt|j|jkr|jrd|_|dSdSdS)z7Resume the writer if queue is below the low water mark.NF)r@r$rErGrCrr r r!rY#rfzAssembler.maybe_resumecCs|jrdSd|_|jdS)z End the stream of frames. Calling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`, or :meth:`put` is safe. They will raise :exc:`EOFError`. NT)rIrEr5rr r r!close.szAssembler.close) r>r?r@r?rArBrCrBrr)rJrKrrL)rJrPrrQr)rJrRrr)rJrKrr^)rJrPrra)rJrRrrb)rWrrrr6) r7r8r9r:r"r r0r_r)rerYrgr r r r!rEs2 "   =    E   ) __future__rrcodecsrcollections.abcrrtypingrrrrr r exceptionsr rEr rrrr__all__getincrementaldecoderrcrrrr r r r!s     0