o g(@sddlmZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z ddl mZddlmZmZmZmZddlmZd gZed Ze d ZGd d d e eZGdd d ZdS)) annotationsN) AsyncIteratorIterable)AnyCallableGenericTypeVar)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)asyncioget_running_looploop get_waiter collectionsdequequeueselfrR/var/www/visachat/venv/lib/python3.10/site-packages/websockets/asyncio/messages.py__init__s zSimpleQueue.__init__intcCs t|jSr)lenrrrrr __len__"s zSimpleQueue.__len__itemrcCs8|j||jdur|js|jddSdSdS)z+Put an item into the queue without waiting.N)rappendrdone set_result)rr%rrr 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)rEOFErrorrr create_futurecancelpopleft)rr*rrr 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)rr2rrr reset9szSimpleQueue.resetcCs0|jdur|js|jtddSdSdS)z8Close the queue, raising EOFError in get() if necessary.Nr,)rr' set_exceptionr-rrrr abort?szSimpleQueue.abortNrr)rr")r%rrr)T)r*r+rr)r2r3rr) __name__ __module__ __qualname____doc__r!r$r)r1r5r7rrrr rs     rc@sleZdZdZddddddfd"ddZd#d$ddZd#d%ddZd&ddZd'ddZd'ddZ d'd d!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. NcCdSrrrrrr YzAssembler.cCr=rrrrrr r>Zr?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@rBrCrEpausedget_in_progressclosed)rr@rBrCrErrr r!Us  zAssembler.__init__decode bool | Nonercs|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).0framerrr sz Assembler.get..)rJr rGr1rK maybe_resumeopcoder r finrCancelledErrorr5r r&joinrL)rrLrRrGrPrrr r1ws8    z Assembler.getAsyncIterator[Data]cCs|jrtdd|_z |j|j IdH}Wn tjy$d|_w||jt us5|jt us5J|dur>|jt u}|rMt }| |j |jVn|j V|js}|j|j IdH}||jtusjJ|rv| |j |jVn|j V|jrTd|_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. rNTNF)rJr rGr1rKrrWrTrUr r UTF8DecoderrLrPrVr )rrLrRdecoderrrr get_iters6   zAssembler.get_iterrRrcCs&|jrtd|j||dS)z Add ``frame`` to the next message. Raises: EOFError: If the stream of frames has ended. r,N)rKr-rGr) maybe_pause)rrRrrr r)s  z Assembler.putcCs>|jdurdSt|j|jkr|jsd|_|dSdSdS)z7Pause the writer if queue is above the high water mark.NT)r@r#rGrIrCrrrr r]  zAssembler.maybe_pausecCs>|jdurdSt|j|jkr|jrd|_|dSdSdS)z7Resume the writer if queue is below the low water mark.NF)rBr#rGrIrErrrr rTr^zAssembler.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)rKrGr7rrrr closeszAssembler.close) r@rArBrArCrDrErDrrr)rLrMrr)rLrMrrY)rRrrrr8) r9r:r;r<r!r1r\r)r]rTr_rrrr rEs " < C   ) __future__rrcodecsrcollections.abcrrtypingrrrr exceptionsr rGr r r rr__all__getincrementaldecoderrZrrrrrrr s    0