i[R%SSKJr SSKrSSKJrJr SSKJrJr SSK J r J r SSK J r SSKJrJr \/\\4rS \S '\/\4rS \S '\"S \S 9r\"S\S 9r"SS5r\ R0"SS\55r\ R0"SS\55r\R6R9SS5\l\R6R9SS5\lSS.S$SjjrS%SjrS&SjrS'Sjr "SS5r!"SS\5r""S S!\5r#S(S"jr$S)S#jr%g)*) annotationsN) AwaitableCallable) TypeAliasTypeVar)_core_util StapledStream) ReceiveStream SendStreamr AsyncHookSyncHook SendStreamT)boundReceiveStreamTcl\rSrSrS SjrS SjrS SjrS SjrSSjrSSjr SSS jjr SSS jjr S r g)_UnboundedByteQueuec[5UlSUl[R"5Ul[ R"S5Ulg)NFz%another task is already fetching data) bytearray_data_closedr ParkingLot_lotr ConflictDetector _fetch_lockselfs IC:\des-py\RoboSAPF\venv\Lib\site-packages\trio/testing/_memory_streams.py__init___UnboundedByteQueue.__init__s8[  $$&  11 3 cFSUlURR5 gNT)rr unpark_allrs r!close_UnboundedByteQueue.close$s  r$cB[5UlUR5 gN)rrr(rs r!close_and_wipe"_UnboundedByteQueue.close_and_wipe(s[  r$cUR(a[R"S5eU=RU- slURR 5 g)Nzvirtual connection closed)rr ClosedResourceErrorrrr'r datas r!put_UnboundedByteQueue.put,s9 <<++,GH H d  r$cZUcg[R"U5nUS:a [S5eg)Nmax_bytes must be >= 1)operatorindex ValueErrorr max_bytess r!_check_max_bytes$_UnboundedByteQueue._check_max_bytes2s1   NN9- q=56 6 r$cUR(dUR(deUc[UR5nUR(a)URSUnURSU2 U(deU$[5$r+)rrlenr)r r;chunks r! _get_impl_UnboundedByteQueue._get_impl9sc||tzz))  DJJI ::JJz *E :I:&L5L; r$NcUR URU5 UR(d!UR(d[R eUR U5sSSS5 $!,(df  g=fr+)rr<rrr WouldBlockrAr:s r! get_nowait_UnboundedByteQueue.get_nowaitEsJ     ! !) ,<< &&&>>),   s AA++ A9ch# UR URU5 UR(d4UR(d#URR 5IShvN O[ R"5IShvN URU5sSSS5 $N<N !,(df  g=f7fr+) rr<rrrparkr checkpointrAr:s r!get_UnboundedByteQueue.getLss     ! !) ,<< iinn&&&&&(((>>),  '(  sA B2AB! B!B!>B?B! B2B!B!! B/+B2)rrrrreturnNoner1zbytes | bytearray | memoryviewrMrN)r; int | NonerMrNr;rPrMrr+) __name__ __module__ __qualname____firstlineno__r"r(r,r2r<rArErJ__static_attributes__r$r!rrs/  7 ---r$rc~\rSrSrSrS S SjjrSSjrSSjrSSjrSSjr SSS jjr SSS jjr S r g)MemorySendStreamVaAn in-memory :class:`~trio.abc.SendStream`. Args: send_all_hook: An async function, or None. Called from :meth:`send_all`. Can do whatever you like. wait_send_all_might_not_block_hook: An async function, or None. Called from :meth:`wait_send_all_might_not_block`. Can do whatever you like. close_hook: A synchronous function, or None. Called from :meth:`close` and :meth:`aclose`. Can do whatever you like. .. attribute:: send_all_hook wait_send_all_might_not_block_hook close_hook All of these hooks are also exposed as attributes on the object, and you can change them at any time. Nc|[R"S5Ul[5UlXlX lX0lg)N!another task is using this stream)r r_conflict_detectorr _outgoing send_all_hook"wait_send_all_might_not_block_hook close_hook)r r_r`ras r!r"MemorySendStream.__init__ls7 #("8"8 /# -.*2T/$r$cX# UR [R"5IShvN [R"5IShvN URR U5 UR bUR 5IShvN SSS5 gNjNON!,(df  g=f7f)zmPlaces the given data into the object's internal buffer, and then calls the :attr:`send_all_hook` (if any). N)r]r rIr^r2r_r0s r!send_allMemorySendStream.send_allzs $ $""$ $ $""$ $ $ NN  t $!!-((*** % $ $ $+ % $P B*BBBB?BBB B*BBB B'#B*cX# UR [R"5IShvN [R"5IShvN URR S5 UR bUR 5IShvN SSS5 gNjNON!,(df  g=f7f)z^Calls the :attr:`wait_send_all_might_not_block_hook` (if any), and then returns immediately. Nr$)r]r rIr^r2r`rs r!wait_send_all_might_not_block.MemorySendStream.wait_send_all_might_not_blocks $ $""$ $ $""$ $ $ NN  s #66B==??? % $ $ $@ % $rfctURR5 URbUR5 gg)zNMarks this stream as closed, and then calls the :attr:`close_hook` (if any). N)r^r(rars r!r(MemorySendStream.closes-  ?? & OO  'r$cj# UR5 [R"5IShvN gN7fz!Same as :meth:`close`, but async.Nr(r rIrs r!acloseMemorySendStream.aclose     )313cT# URRU5IShvN $N7f)axRetrieves data from the internal buffer, blocking if necessary. Args: max_bytes (int or None): The maximum amount of data to retrieve. None (the default) means to retrieve all the data that's present (but still blocks until at least one byte is available). Returns: If this stream has been closed, an empty bytearray. Otherwise, the requested data. N)r^rJr:s r!get_dataMemorySendStream.get_datas"^^'' 2222 (&(c8URRU5$)zRetrieves data from the internal buffer, but doesn't block. See :meth:`get_data` for details. Raises: trio.WouldBlock: if no data is available to retrieve. )r^rEr:s r!get_data_nowait MemorySendStream.get_data_nowaits~~((33r$)r]r^rar_r`)NNN)r_AsyncHook | Noner`rzraSyncHook | NonerMrNrOrLr+rQ) rRrSrTrU__doc__r"rdrhr(rortrxrVrWr$r!rYrYVsa,+/?C&* %' %-= %$ %  % + @"! 3 4 4r$rYcj\rSrSrSrS S SjjrS SSjjrSSjrSSjrSSjr SS jr S r g)MemoryReceiveStreamaAn in-memory :class:`~trio.abc.ReceiveStream`. Args: receive_some_hook: An async function, or None. Called from :meth:`receive_some`. Can do whatever you like. close_hook: A synchronous function, or None. Called from :meth:`close` and :meth:`aclose`. Can do whatever you like. .. attribute:: receive_some_hook close_hook Both hooks are also exposed as attributes on the object, and you can change them at any time. Nc~[R"S5Ul[5UlSUlXlX lg)Nr\F)r rr]r _incomingrreceive_some_hookra)r rras r!r"MemoryReceiveStream.__init__s8 #("8"8 /# -. !2$r$c# UR [R"5IShvN [R"5IShvN UR(a[ReUR bUR 5IShvN UR RU5IShvN nUR(a[ReUsSSS5 $NNNWN6!,(df  g=f7f)zxCalls the :attr:`receive_some_hook` (if any), and then retrieves data from the internal buffer, blocking if necessary. N)r]r rIrr/rrrJ)r r;r1s r! receive_some MemoryReceiveStream.receive_somes $ $""$ $ $""$ $ $||///%%1,,... ++I66D||///% $ $ $/ 7% $sc C8C'CC'C!AC' C# "C'.C%/&C' C8C'!C'#C'%C'' C51C8cSUlURR5 URbUR5 gg)zVDiscards any pending data from the internal buffer, and marks this stream as closed. TN)rrr,rars r!r(MemoryReceiveStream.closes4   %%' ?? & OO  'r$cj# UR5 [R"5IShvN gN7frmrnrs r!roMemoryReceiveStream.aclose rqrrc:URRU5 g)z.Appends the given data to the internal buffer.N)rr2r0s r!put_dataMemoryReceiveStream.put_datas 4 r$c8URR5 g)z2Adds an end-of-file marker to the internal buffer.N)rr(rs r!put_eofMemoryReceiveStream.put_eofs r$)rr]rrar)NN)rrzrar{rMrNr+rQrLrO) rRrSrTrUr|r"rr(rorrrVrWr$r!r~r~sI$/3&* %+ %$ %  %.! !r$r~z._memory_streams)r;cURU5nU(dUR5 gUR U5 g![Ra gf=f![R a [R "S5Sef=f)aTake data out of the given :class:`MemorySendStream`'s internal buffer, and put it into the given :class:`MemoryReceiveStream`'s internal buffer. Args: memory_send_stream (MemorySendStream): The stream to get data from. memory_receive_stream (MemoryReceiveStream): The stream to put data into. max_bytes (int or None): The maximum amount of data to transfer in this call, or None to transfer all available data. Returns: True if it successfully transferred some data, or False if there was no data to transfer. This is used to implement :func:`memory_stream_one_way_pair` and :func:`memory_stream_pair`; see the latter's docstring for an example of how you might use it yourself. FzMemoryReceiveStream was closedNT)rxr rDrrr/BrokenResourceError)memory_send_streammemory_receive_streamr;r1s r!memory_stream_pumpr s0!11)<T ! ) ) +  " * *4 0      $ $T''(HItSTs?AAAA,Bcv^^^[5m[5mSUU4SjjmSU4SjjnUTlTTlTT4$)u Create a connected, pure-Python, unidirectional stream with infinite buffering and flexible configuration options. You can think of this as being a no-operating-system-involved Trio-streamsified version of :func:`os.pipe` (except that :func:`os.pipe` returns the streams in the wrong order – we follow the superior convention that data flows from left to right). Returns: A tuple (:class:`MemorySendStream`, :class:`MemoryReceiveStream`), where the :class:`MemorySendStream` has its hooks set up so that it calls :func:`memory_stream_pump` from its :attr:`~MemorySendStream.send_all_hook` and :attr:`~MemorySendStream.close_hook`. The end result is that data automatically flows from the :class:`MemorySendStream` to the :class:`MemoryReceiveStream`. But you're also free to rearrange things however you like. For example, you can temporarily set the :attr:`~MemorySendStream.send_all_hook` to None if you want to simulate a stall in data transmission. Or see :func:`memory_stream_pair` for a more elaborate example. c>[TT5 gr+)r) recv_stream send_streamsr!$pump_from_send_stream_to_recv_streamHmemory_stream_one_way_pair..pump_from_send_stream_to_recv_streamas ; 4r$c># T"5 g7fr+rW)rsr!*async_pump_from_send_stream_to_recv_streamNmemory_stream_one_way_pair..async_pump_from_send_stream_to_recv_streames ,.s rL)rYr~r_ra)rrrrs @@@r!memory_stream_one_way_pairrFsB0#$K%'K55/!KKAK  ##r$cXU"5upU"5up4[X5n[X25nXV4$r+r ) one_way_pair pipe1_send pipe1_recv pipe2_send pipe2_recvstream1stream2s r!_make_stapled_pairrms5 *^J)^JJ3GJ3G  r$c [[5$)a Create a connected, pure-Python, bidirectional stream with infinite buffering and flexible configuration options. This is a convenience function that creates two one-way streams using :func:`memory_stream_one_way_pair`, and then uses :class:`~trio.StapledStream` to combine them into a single bidirectional stream. This is like a no-operating-system-involved, Trio-streamsified version of :func:`socket.socketpair`. Returns: A pair of :class:`~trio.StapledStream` objects that are connected so that data automatically flows from one to the other in both directions. After creating a stream pair, you can send data back and forth, which is enough for simple tests:: left, right = memory_stream_pair() await left.send_all(b"123") assert await right.receive_some() == b"123" await right.send_all(b"456") assert await left.receive_some() == b"456" But if you read the docs for :class:`~trio.StapledStream` and :func:`memory_stream_one_way_pair`, you'll see that all the pieces involved in wiring this up are public APIs, so you can adjust to suit the requirements of your tests. For example, here's how to tweak a stream so that data flowing from left to right trickles in one byte at a time (but data flowing from right to left proceeds at full speed):: left, right = memory_stream_pair() async def trickle(): # left is a StapledStream, and left.send_stream is a MemorySendStream # right is a StapledStream, and right.recv_stream is a MemoryReceiveStream while memory_stream_pump(left.send_stream, right.recv_stream, max_bytes=1): # Pause between each byte await trio.sleep(1) # Normally this send_all_hook calls memory_stream_pump directly without # passing in a max_bytes. We replace it with our custom version: left.send_stream.send_all_hook = trickle And here's a simple test using our modified stream objects:: async def sender(): await left.send_all(b"12345") await left.send_eof() async def receiver(): async for data in right: print(data) async with trio.open_nursery() as nursery: nursery.start_soon(sender) nursery.start_soon(receiver) By default, this will print ``b"12345"`` and then immediately exit; with our trickle stream it instead sleeps 1 second, then prints ``b"1"``, then sleeps 1 second, then prints ``b"2"``, etc. Pro-tip: you can insert sleep calls (like in our example above) to manipulate the flow of data across tasks... and then use :class:`MockClock` and its :attr:`~MockClock.autojump_threshold` functionality to keep your test suite running quickly. If you want to stress test a protocol implementation, one nice trick is to use the :mod:`random` module (preferably with a fixed seed) to move random numbers of bytes at a time, and insert random sleeps in between them. You can also set up a custom :attr:`~MemoryReceiveStream.receive_some_hook` if you want to manipulate things on the receiving side, and not just the sending side. )rrrWr$r!memory_stream_pairrzsZ 8 99r$ch\rSrSrS SjrS SjrS SjrS SjrS SjrSSjr S Sjr SSS jjr S r g )_LockstepByteQueueic[5UlSUlSUlSUl[ R "5Ul[R"S5Ul [R"S5Ul g)NFzanother task is already sendingz!another task is already receiving) rr_sender_closed_receiver_closed_receiver_waitingr r_waitersr r_send_conflict_detector_receive_conflict_detectorrs r!r"_LockstepByteQueue.__init__sa[ # %!&((* ','='= -( $+0*@*@ /+ 'r$c8URR5 gr+)rr'rs r!_something_happened&_LockstepByteQueue._something_happeneds   "r$c# U"5(aOGUR(dUR(aO$URR5IShvN MT[R "5IShvN gN$N7fr+)rrrrHr rI)r fns r! _wait_for_LockstepByteQueue._wait_fors]tt""d&;&;--$$& & &      ' s$AA:A6A:0A81A:8A:c2SUlUR5 gr&)rrrs r! close_sender_LockstepByteQueue.close_senders"   "r$c2SUlUR5 gr&)rrrs r!close_receiver!_LockstepByteQueue.close_receivers $   "r$c2^# TR TR(a[ReTR(a[R eTR (aeT=R U- slTR5 TRU4Sj5IShvN TR(a[ReTR (a!TR(a[R eSSS5 gN`!,(df  g=f7f)Nc">TRS:H$Nr$rrsr!-_LockstepByteQueue.send_all..ss):r$) rrr r/rrrrrr0s` r!rd_LockstepByteQueue.send_alls  ) )""///$$///zz !> JJ$ J  $ $ &..!:; ; ;""///zzd33///* ) <* )s0 DBD#D$AD; DD DDc^# TR TR(a[ReTR(a'[R "5IShvN SSS5 gTR U4Sj5IShvN TR(a[ReSSS5 gNVN1!,(df  g=f7f)Nc>TR$r+)rrsr!rB_LockstepByteQueue.wait_send_all_might_not_block.. s )?)?r$)rrr r/rrIrrs`r!rh0_LockstepByteQueue.wait_send_all_might_not_blocks  ) )""///$$&&((( * ) ..!?@ @ @""///#* )) @ * )sM CA B6B2B6! C*B6B4%B6) C2B64B66 CCNc^# TR Ub'[R"U5nUS:a [S5eTR(a[ R eSTlTR5 TRU4Sj5IShvN STlTR(a[ R eTR(a9TRSUnTRSU2 TR5 UsSSS5 $TR(deSSS5 gN!STlf=f!,(df  g=f7f)Nr5r6Tc">TRS:g$rrrsr!r1_LockstepByteQueue.receive_some..sTZZ3->r$Fr$) rr7r8r9rr r/rrrrr)r r;gots` r!r_LockstepByteQueue.receive_some s  , ,$$NN95 q=$%=>>$$///%)D "  $ $ & /nn%>???).&$$///zzjj),JJz z*((*3- ,6****9- ,@).&- ,sT D?A#D.4D" D  D"A(D.9 D?D. D? D"" D++D.. D<8D?)rrrrrrrrL)rzCallable[[], bool]rMrNrOr+r;rPrMzbytes | bytearray) rRrSrTrUr"rrrrrdrhrrVrWr$r!rrs/  # !## 0 0r$rcF\rSrSrS SjrS SjrS SjrS SjrS SjrSr g) _LockstepSendStreami-cXlgr+_lbqr lbqs r!r"_LockstepSendStream.__init__. r$c8URR5 gr+)rrrs r!r(_LockstepSendStream.close1s  r$cj# UR5 [R"5IShvN gN7fr+rnrs r!ro_LockstepSendStream.aclose4     rrcV# URRU5IShvN gN7fr+)rrdr0s r!rd_LockstepSendStream.send_all8sii  &&&s )')cT# URR5IShvN gN7fr+)rrhrs r!rh1_LockstepSendStream.wait_send_all_might_not_block;sii55777s (&(rNrrrMrNrLrO) rRrSrTrUr"r(rordrhrVrWr$r!rr-s!!'8r$rc@\rSrSrSSjrS SjrS SjrS S SjjrSrg) _LockstepReceiveStreami?cXlgr+rrs r!r"_LockstepReceiveStream.__init__@rr$c8URR5 gr+)rrrs r!r(_LockstepReceiveStream.closeCs   "r$cj# UR5 [R"5IShvN gN7fr+rnrs r!ro_LockstepReceiveStream.acloseFrrrNcT# URRU5IShvN $N7fr+)rrr:s r!r#_LockstepReceiveStream.receive_someJs YY++I6666rvrrrLr+r) rRrSrTrUr"r(rorrVrWr$r!rr?s#!77r$rcB[5n[U5[U54$)aCreate a connected, pure Python, unidirectional stream where data flows in lockstep. Returns: A tuple (:class:`~trio.abc.SendStream`, :class:`~trio.abc.ReceiveStream`). This stream has *absolutely no* buffering. Each call to :meth:`~trio.abc.SendStream.send_all` will block until all the given data has been returned by a call to :meth:`~trio.abc.ReceiveStream.receive_some`. This can be useful for testing flow control mechanisms in an extreme case, or for setting up "clogged" streams to use with :func:`check_one_way_stream` and friends. In addition to fulfilling the :class:`~trio.abc.SendStream` and :class:`~trio.abc.ReceiveStream` interfaces, the return objects also have a synchronous ``close`` method. )rrr)rs r!lockstep_stream_one_way_pairrNs".  C s #%;C%@ @@r$c [[5$)asCreate a connected, pure-Python, bidirectional stream where data flows in lockstep. Returns: A tuple (:class:`~trio.StapledStream`, :class:`~trio.StapledStream`). This is a convenience function that creates two one-way streams using :func:`lockstep_stream_one_way_pair`, and then uses :class:`~trio.StapledStream` to combine them into a single bidirectional stream. )rrrWr$r!lockstep_stream_pairris : ;;r$)rrYrr~r;rPrMbool)rMz,tuple[MemorySendStream, MemoryReceiveStream])rz0Callable[[], tuple[SendStreamT, ReceiveStreamT]]rMz]tuple[StapledStream[SendStreamT, ReceiveStreamT], StapledStream[SendStreamT, ReceiveStreamT]])rMzqtuple[StapledStream[MemorySendStream, MemoryReceiveStream], StapledStream[MemorySendStream, MemoryReceiveStream]])rMz tuple[SendStream, ReceiveStream])rMzYtuple[StapledStream[SendStream, ReceiveStream], StapledStream[SendStream, ReceiveStream]])& __future__rr7collections.abcrrtypingrrrr r _highlevel_genericr abcr robjectr__annotations__rrrrfinalrYr~rSreplacerrrrrrrrrrWr$r!rsu"/%.+If$5 56 96r6z*)*m:6 )?<-<-~o4zo4 o4dJ-J J\/99AA"5!?!?!G!G"! #(#.# #  #L$$N B  M:j[[|8*8$ 7] 7A6