a њh?1@sddlmZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z ddlmZmZmZmZddlmZdd lmZd gZed ZGd d d ZdS) ) annotationsN)AnyCallableIterableIteratorLiteraloverload)ConcurrencyError) OP_BINARYOP_CONTOP_TEXTFrame)Data)Deadline Assemblerzutf-8c@sleZdZdZddddddfdddddd d d Zd6d d dddZdddddZed dddddZed dddddZed7d dddddZed8d dddddZed9d dd dd!dZd:d dd dd"dZedd#d$d%d&Z edd'd$d(d&Z ed;dd)d$d*d&Z d&zAssembler.cCsdSrrrrrrr'rz int | NonezCallable[[], Any]None)highlowpauseresumereturncCst|_t|_|dur,|dur,|d}|durD|durD|d}|durt|durt|dkrdtd||krttd|||_|_||_ ||_ d|_ d|_ d|_ dS)Nrz%low must be positive or equal to zeroz)high must be greater than or equal to lowF) threadingLockmutexqueue SimpleQueueframes ValueErrorrrrrpausedget_in_progressclosed)selfrrrrrrr__init__"s"  zAssembler.__init__z float | Noner)timeoutrcCs|jr:z|jjdd}Wqtjy6tddYq0n^z4|dur\|dkr\|jjdd}n|jjd|d}Wn(tjytd|dd dYn0|durtd|S) NFblockstream of frames endedrT)r-r+z timed out in z.1fs)r(r$getr"EmptyEOFError TimeoutError)r)r+framerrrget_next_frameHszAssembler.get_next_framezIterable[Frame])r$rc Cs|jpg}z||jjddqWntjy:Yn0|D]}|j|q@|D]}|j|qVWdn1s|0YdS)NFr,)r!appendr$r0r"r1put)r)r$Zqueuedr4rrr reset_queue^szAssembler.reset_queuez Literal[True]str)r+decodercCsdSrrr)r+r:rrrr0tsz Assembler.getzLiteral[False]bytescCsdSrrr;rrrr0wscCsdSrrr;rrrr0zscCsdSrrr;rrrr0}sz bool | NonercCsdSrrr;rrrr0scCs||j$|jrtdd|_Wdn1s00Yz t|}||jdd}|j|Wdn1s|0Y|jtus|jt usJ|dur|jtu}|g}|j s@z||jdd}Wnt y| |Yn0|j|Wdn1s0Y|jt us4J||qWd|_nd|_0ddd|D}|rt|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: timeout: If a timeout is provided and elapses before a complete message is received, :meth:`get` raises :exc:`TimeoutError`. 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. TimeoutError: If a timeout is provided and elapses before a complete message is received. &get() or get_iter() is already runningTNF)Zraise_if_elapsedrcss|] }|jVqdSr)data).0r4rrr rz Assembler.get..)r!r'r rr5r+ maybe_resumeopcoder r finr3r8r r6joinr:)r)r+r:deadliner4r$r>rrrr0s<$&    (z Iterator[str])r:rcCsdSrrr)r:rrrget_iterszAssembler.get_iterzIterator[bytes]cCsdSrrrFrrrrGszIterator[Data]cCsdSrrrFrrrrGsccs<|j$|jrtdd|_Wdn1s00Y|}|j|Wdn1sf0Y|jtus|jtusJ|dur|jtu}|rt}| |j |j Vn|j V|j s2|}|j|Wdn1s0Y|jt usJ|r(| |j |j Vq|j Vqd|_dS)a Stream the next message. Iterating the return value of :meth:`get_iter` 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. r=TNF) r!r'r r5rArBr r UTF8Decoderr:r>rCr )r)r:r4decoderrrrrGs.$ & & )r4rcCsL|j2|jrtd|j||Wdn1s>0YdS)z Add ``frame`` to the next message. Raises: EOFError: If the stream of frames has ended. r.N)r!r(r2r$r7 maybe_pause)r)r4rrrr7s  z Assembler.put)rcCsD|jdurdS|jsJ|j|jkr@|js@d|_|dS)z7Pause the writer if queue is above the high water mark.NT)rr!lockedr$qsizer&rr)rrrrJ*s  zAssembler.maybe_pausecCsD|jdurdS|jsJ|j|jkr@|jr@d|_|dS)z7Resume the writer if queue is below the low water mark.NF)rr!rKr$rLr&rrMrrrrA7s  zAssembler.maybe_resumecCsl|jR|jrWddSd|_|jr6|jd|jrJd|_|Wdn1s^0YdS)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`. NTF)r!r(r'r$r7r&rrMrrrcloseDs zAssembler.close)N)N)N)NN)NN)N)N)__name__ __module__ __qualname____doc__r*r5r8rr0rGr7rJrArNrrrrrs<&G>  ) __future__rcodecsr"rtypingrrrrrr exceptionsr r$r r r rrutilsr__all__getincrementaldecoderrHrrrrrs