o ͷ6i[@stUddlmZddlZddlmZmZddlmZmZddl m Z m Z ddl m Z ddlmZmZegeefZd ed <egefZd ed <ed ed Zeded ZGdddZe jGdddeZe jGdddeZejdde_ejdde_ddd7ddZd8d!d"Zd9d&d'Zd:d)d*Z Gd+d,d,Z!Gd-d.d.eZ"Gd/d0d0eZ#d;d2d3Z$d= 1)operatorindex ValueErrorrr,rrr _check_max_bytes2s  z$_UnboundedByteQueue._check_max_bytesrcCsT|js|jsJ|durt|j}|jr'|jd|}|jd|=|s%J|StSr%)rrlenr)rr,chunkrrr _get_impl9s  z_UnboundedByteQueue._get_implNcCsP|j|||js|jstj||WdS1s!wYdSr%)rr4rrr WouldBlockr7r3rrr get_nowaitEs   $z_UnboundedByteQueue.get_nowaitcsl|j(|||js|js|jIdHntIdH||WdS1s/wYdSr%) rr4rrrparkr checkpointr7r3rrr getLs  $z_UnboundedByteQueue.getrrr'r(rr)r,r-rrr,r-rrr%) __name__ __module__ __qualname__r!r$r&r+r4r7r9r<rrrr rs      rc@sbeZdZdZ   ddd d ZdddZd ddZd ddZd ddZd!d"ddZ d!d"ddZ dS)#MemorySendStreamaAn 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. N send_all_hookAsyncHook | None"wait_send_all_might_not_block_hook close_hookSyncHook | NonerrcCs*td|_t|_||_||_||_dS)N!another task is using this stream)r r_conflict_detectorr _outgoingrDrFrG)rrDrFrGrrr r!ls zMemorySendStream.__init__r'r(cs~|j1tIdHtIdH|j||jdur-|IdHWddSWddS1s8wYdS)z}Places the given data into the object's internal buffer, and then calls the :attr:`send_all_hook` (if any). N)rJrr;rKr+rDr*rrr send_allzs  "zMemorySendStream.send_allcs~|j1tIdHtIdH|jd|jdur-|IdHWddSWddS1s8wYdS)znCalls the :attr:`wait_send_all_might_not_block_hook` (if any), and then returns immediately. N)rJrr;rKr+rFrrrr wait_send_all_might_not_blocks  "z.MemorySendStream.wait_send_all_might_not_blockcCs$|j|jdur|dSdS)z^Marks this stream as closed, and then calls the :attr:`close_hook` (if any). N)rKr$rGrrrr r$s  zMemorySendStream.closec|tIdHdSz!Same as :meth:`close`, but async.Nr$rr;rrrr aclosezMemorySendStream.acloser,r-rcs|j|IdHS)aRetrieves 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)rKr<r3rrr get_dataszMemorySendStream.get_datacCs |j|S)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. )rKr9r3rrr get_data_nowaits z MemorySendStream.get_data_nowait)NNN)rDrErFrErGrHrrr>r=r%r?) r@rArB__doc__r!rLrNr$rRrTrUrrrr rCVs     rCc@sTeZdZdZ  ddd d ZddddZdddZdddZdddZdddZ dS) 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. Nreceive_some_hookrErGrHrrcCs*td|_t|_d|_||_||_dS)NrIF)r rrJr _incomingrrXrG)rrXrGrrr r!s zMemoryReceiveStream.__init__r,r-rcs|j9tIdHtIdH|jrtj|jdur%|IdH|j|IdH}|jr4tj|WdS1s@wYdS)zCalls the :attr:`receive_some_hook` (if any), and then retrieves data from the internal buffer, blocking if necessary. N)rJrr;rr)rXrYr<)rr,r'rrr receive_somes $z MemoryReceiveStream.receive_somecCs*d|_|j|jdur|dSdS)zfDiscards any pending data from the internal buffer, and marks this stream as closed. TN)rrYr&rGrrrr r$s    zMemoryReceiveStream.closecrOrPrQrrrr rR rSzMemoryReceiveStream.acloser'r(cCs|j|dS)z.Appends the given data to the internal buffer.N)rYr+r*rrr put_dataszMemoryReceiveStream.put_datacCs|jdS)z2Adds an end-of-file marker to the internal buffer.N)rYr$rrrr put_eofszMemoryReceiveStream.put_eof)NN)rXrErGrHrrr%r?r=r>) r@rArBrVr!rZr$rRr[r\rrrr rWs  rWz._memory_streams)r,memory_send_streammemory_receive_streamr,r-rboolcCsfz||}Wn tjyYdSwz|s|WdS||WdStjy2tddw)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)rUrr8r\r[r)BrokenResourceError)r^r_r,r'rrr memory_stream_pump s   rb,tuple[MemorySendStream, MemoryReceiveStream]cs>ttdfdd dfdd }|__fS) uQCreate 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. rrcstdSr%)rbr) recv_stream send_streamrr $pump_from_send_stream_to_recv_streamazHmemory_stream_one_way_pair..pump_from_send_stream_to_recv_streamcs dSr%rr)rfrr *async_pump_from_send_stream_to_recv_streames zNmemory_stream_one_way_pair..async_pump_from_send_stream_to_recv_streamNr=)rCrWrDrG)rhr)rfrdrer memory_stream_one_way_pairFsri one_way_pair0Callable[[], tuple[SendStreamT, ReceiveStreamT]]]tuple[StapledStream[SendStreamT, ReceiveStreamT], StapledStream[SendStreamT, ReceiveStreamT]]cCs0|\}}|\}}t||}t||}||fSr%r )rj pipe1_send pipe1_recv pipe2_send pipe2_recvstream1stream2rrr _make_stapled_pairms    rsqtuple[StapledStream[MemorySendStream, MemoryReceiveStream], StapledStream[MemorySendStream, MemoryReceiveStream]]cCttS)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. )rsrirrrr memory_stream_pairzsMrvc@s^eZdZdddZdddZdd d Zdd d Zdd dZdddZdddZ ddddZ dS) _LockstepByteQueuerrcCs@t|_d|_d|_d|_t|_t d|_ t d|_ dS)NFzanother task is already sendingz!another task is already receiving) rr_sender_closed_receiver_closed_receiver_waitingrr_waitersr r_send_conflict_detector_receive_conflict_detectorrrrr r!s  z_LockstepByteQueue.__init__cC|jdSr%)r{r#rrrr _something_happenedrgz&_LockstepByteQueue._something_happenedfnCallable[[], bool]cs> |rn|js |jr n |jIdHqtIdHdSr%)rxryr{r:rr;)rrrrr _wait_fors z_LockstepByteQueue._wait_forcCd|_|dSr")rxrrrrr close_sender z_LockstepByteQueue.close_sendercCrr")ryrrrrr close_receiverrz!_LockstepByteQueue.close_receiverr'r(csjHjr tjjrtjjrJj|7_fddIdHjr3tjjr<jrDtjWddSWddS1sOwYdS)Ncs jdkSNrMrrrrr  z-_LockstepByteQueue.send_all..) r|rxrr)ryrarrrr*rrr rLs$   "z_LockstepByteQueue.send_allcsj4jr tjjrtIdH WddSfddIdHjr0tjWddS1s;wYdS)NcsjSr%)rzrrrr r szB_LockstepByteQueue.wait_send_all_might_not_block..)r|rxrr)ryr;rrrrr rNs"z0_LockstepByteQueue.wait_send_all_might_not_blockNr,r-bytes | bytearrayc sjf|durt|}|dkrtdjrtjd_z fddIdHWd_nd_wjr?tjj r\j d|}j d|=|WdSj saJ WddS1smwYdS)Nr.r/Tcs jdkSrrrrrr rrz1_LockstepByteQueue.receive_some..FrM) r}r0r1r2ryrr)rzrrrrx)rr,gotrrr rZ s0   $z_LockstepByteQueue.receive_somer=)rrrrr>r%r,r-rr) r@rArBr!rrrrrLrNrZrrrr rws      rwc@s>eZdZdddZdddZdd d Zdd dZdddZdS)_LockstepSendStreamlbqrwrrcC ||_dSr%_lbqrrrrr r!. z_LockstepSendStream.__init__cCr~r%)rrrrrr r$1rgz_LockstepSendStream.closec|tIdHdSr%rQrrrr rR4z_LockstepSendStream.acloser'r(cs|j|IdHdSr%)rrLr*rrr rL8sz_LockstepSendStream.send_allcs|jIdHdSr%)rrNrrrr rN;sz1_LockstepSendStream.wait_send_all_might_not_blockNrrwrrr=r>)r@rArBr!r$rRrLrNrrrr r-s     rc@s6eZdZdddZdddZdd d ZddddZd S)_LockstepReceiveStreamrrwrrcCrr%rrrrr r!@rz_LockstepReceiveStream.__init__cCr~r%)rrrrrr r$Crgz_LockstepReceiveStream.closecrr%rQrrrr rRFrz_LockstepReceiveStream.acloseNr,r-rcs|j|IdHSr%)rrZr3rrr rZJsz#_LockstepReceiveStream.receive_somerr=r%r)r@rArBr!r$rRrZrrrr r?s    r tuple[SendStream, ReceiveStream]cCst}t|t|fS)a Create 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. )rwrr)rrrr lockstep_stream_one_way_pairNsrYtuple[StapledStream[SendStream, ReceiveStream], StapledStream[SendStream, ReceiveStream]]cCru)aCreate 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. )rsrrrrr lockstep_stream_pairisr)r^rCr_rWr,r-rr`)rrc)rjrkrrl)rrt)rr)rr)& __future__rr0collections.abcrrtypingrrr]rr _highlevel_genericr abcr r objectr__annotations__rrrrfinalrCrWrAreplacerbrirsrvrwrrrrrrrr s@   ?rN  & ' U^