a xdfS@sddlZddlmZddlmZddlmZddlmZmZGdddZ Gd d d eej d Z Gd d d eej d Z ddddZ ddZddZddZGdddZGdddeZGdddeZddZdd ZdS)!N)_core StapledStream)_util) SendStream ReceiveStreamc@sPeZdZddZddZddZddZd d Zd d ZdddZ dddZ d S)_UnboundedByteQueuecCs(t|_d|_t|_td|_dS)NFz%another task is already fetching data) bytearray_data_closedr ParkingLot_lotrConflictDetector _fetch_lockselfrT/var/www/html/Ranjet/env/lib/python3.9/site-packages/trio/testing/_memory_streams.py__init__s  z_UnboundedByteQueue.__init__cCsd|_|jdSNT)r r unpark_allrrrrclosesz_UnboundedByteQueue.closecCst|_|dSN)r r rrrrrclose_and_wipesz"_UnboundedByteQueue.close_and_wipecCs,|jrtd|j|7_|jdS)Nzvirtual connection closed)r rClosedResourceErrorr rrrdatarrrput"s z_UnboundedByteQueue.putcCs*|dur dSt|}|dkr&tddS)Nmax_bytes must be >= 1)operatorindex ValueErrorr max_bytesrrr_check_max_bytes(s  z$_UnboundedByteQueue._check_max_bytescCsX|js|jsJ|dur"t|j}|jrN|jd|}|jd|=|sJJ|StSdSr)r r lenr )rr%chunkrrr _get_impl/s  z_UnboundedByteQueue._get_implNcCsP|j6|||js$|js$tj||WdS1sB0YdSr)rr&r r r WouldBlockr)r$rrr get_nowait;s   z_UnboundedByteQueue.get_nowaitcsj|jP|||js0|js0|jIdHntIdH||WdS1s\0YdSr) rr&r r rparkr checkpointr)r$rrrgetBs   z_UnboundedByteQueue.get)N)N) __name__ __module__ __qualname__rrrrr&r)r+r.rrrrr s  r c@sNeZdZdZdddZddZddZd d Zd d Zdd dZ dddZ 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. NcCs*td|_t|_||_||_||_dS)N!another task is using this stream)rr_conflict_detectorr _outgoing send_all_hook"wait_send_all_might_not_block_hook close_hook)rr6r7r8rrrraszMemorySendStream.__init__csj|jPtIdHtIdH|j||jdurH|IdHWdn1s\0YdS)z}Places the given data into the object's internal buffer, and then calls the :attr:`send_all_hook` (if any). N)r4rr-r5rr6rrrrsend_allos   zMemorySendStream.send_allcsj|jPtIdHtIdH|jd|jdurH|IdHWdn1s\0YdS)znCalls the :attr:`wait_send_all_might_not_block_hook` (if any), and then returns immediately. N)r4rr-r5rr7rrrrwait_send_all_might_not_block}s   z.MemorySendStream.wait_send_all_might_not_blockcCs |j|jdur|dS)z^Marks this stream as closed, and then calls the :attr:`close_hook` (if any). N)r5rr8rrrrrs  zMemorySendStream.closecs|tIdHdSz!Same as :meth:`close`, but async.Nrrr-rrrracloseszMemorySendStream.aclosecs|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)r5r.r$rrrget_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. )r5r+r$rrrget_data_nowaits z MemorySendStream.get_data_nowait)NNN)N)N) r/r0r1__doc__rr9r;rr>r?r@rrrrr2Ls  r2) metaclassc@sDeZdZdZdddZdddZddZd d Zd d Zd dZ 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. NcCs*td|_t|_d|_||_||_dS)Nr3F)rrr4r _incomingr receive_some_hookr8)rrEr8rrrrszMemoryReceiveStream.__init__cs|jrtIdHtIdH|jr0tj|jdurH|IdH|j|IdH}|jrftj|WdS1s~0YdS)zCalls the :attr:`receive_some_hook` (if any), and then retrieves data from the internal buffer, blocking if necessary. N)r4rr-r rrErDr.)rr%rrrr receive_somes z MemoryReceiveStream.receive_somecCs&d|_|j|jdur"|dS)zfDiscards any pending data from the internal buffer, and marks this stream as closed. TN)r rDrr8rrrrrs  zMemoryReceiveStream.closecs|tIdHdSr<r=rrrrr>szMemoryReceiveStream.aclosecCs|j|dS)z.Appends the given data to the internal buffer.N)rDrrrrrput_dataszMemoryReceiveStream.put_datacCs|jdS)z2Adds an end-of-file marker to the internal buffer.N)rDrrrrrput_eofszMemoryReceiveStream.put_eof)NN)N) r/r0r1rArrFrr>rGrHrrrrrCs  rC)r%cCsfz||}Wntjy$YdS0z|s6|n ||Wntjy`tdYn0dS)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 closedT)r@rr*rHrGrBrokenResourceError)Zmemory_send_streamZmemory_receive_streamr%rrrrmemory_stream_pumps rJcs:ttfddfdd}|__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. cstdSr)rJr) recv_stream send_streamrr$pump_from_send_stream_to_recv_streamCszHmemory_stream_one_way_pair..pump_from_send_stream_to_recv_streamcs dSrrr)rMrr*async_pump_from_send_stream_to_recv_streamFszNmemory_stream_one_way_pair..async_pump_from_send_stream_to_recv_stream)r2rCr6r8)rNr)rMrKrLrmemory_stream_one_way_pair(s rOcCs0|\}}|\}}t||}t||}||fSrr)Z one_way_pairZ pipe1_sendZ pipe1_recvZ pipe2_sendZ pipe2_recvZstream1Zstream2rrr_make_stapled_pairNs     rPcCsttS)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. )rPrOrrrrmemory_stream_pairVsJrQc@sNeZdZddZddZddZddZd d Zd d Zd dZ dddZ dS)_LockstepByteQueuecCs@t|_d|_d|_d|_t|_t d|_ t d|_ dS)NFzanother task is already sendingz!another task is already receiving) r r _sender_closed_receiver_closed_receiver_waitingrr _waitersrr_send_conflict_detector_receive_conflict_detectorrrrrrs z_LockstepByteQueue.__init__cCs|jdSr)rVrrrrr_something_happenedsz&_LockstepByteQueue._something_happenedcs:|rq(|js(|jrq(|jIdHqtIdHdSr)rSrTrVr,rr-)rfnrrr _wait_fors  z_LockstepByteQueue._wait_forcCsd|_|dSr)rSrYrrrr close_sendersz_LockstepByteQueue.close_sendercCsd|_|dSr)rTrYrrrrclose_receiversz!_LockstepByteQueue.close_receivercsj~jrtjjr tjjr*Jj|7_fddIdHjrdtjjrvjrvtjWdn1s0YdS)Ncsj Srr rrrrr:z-_LockstepByteQueue.send_all..) rWrSrrrTrIr rYr[rrrrr9s  z_LockstepByteQueue.send_allcs~jdjrtjjr8tIdHWddSfddIdHjr\tjWdn1sp0YdS)NcsjSr)rUrrrrr_r:zB_LockstepByteQueue.wait_send_all_might_not_block..)rWrSrrrTr-r[rrrrr;sz0_LockstepByteQueue.wait_send_all_might_not_blockNc sj|dur*t|}|dkr*tdjr6tjd_z" fddIdHWd_nd_0jr|tjj rj d|}j d|=|WdSj sJWddSWdn1s0YdS)Nrr TcsjSrr^rrrrr_r:z1_LockstepByteQueue.receive_some..Fr:) rXr!r"r#rTrrrUrYr[r rS)rr%gotrrrrFs*   z_LockstepByteQueue.receive_some)N) r/r0r1rrYr[r\r]r9r;rFrrrrrRs   rRc@s4eZdZddZddZddZddZd d Zd S) _LockstepSendStreamcCs ||_dSr_lbqrlbqrrrrsz_LockstepSendStream.__init__cCs|jdSr)rcr\rrrrr sz_LockstepSendStream.closecs|tIdHdSrr=rrrrr> sz_LockstepSendStream.aclosecs|j|IdHdSr)rcr9rrrrr9sz_LockstepSendStream.send_allcs|jIdHdSr)rcr;rrrrr;sz1_LockstepSendStream.wait_send_all_might_not_blockN)r/r0r1rrr>r9r;rrrrras rac@s.eZdZddZddZddZd dd ZdS) _LockstepReceiveStreamcCs ||_dSrrbrdrrrrsz_LockstepReceiveStream.__init__cCs|jdSr)rcr]rrrrrsz_LockstepReceiveStream.closecs|tIdHdSrr=rrrrr>sz_LockstepReceiveStream.acloseNcs|j|IdHSr)rcrFr$rrrrF#sz#_LockstepReceiveStream.receive_some)N)r/r0r1rrr>rFrrrrrfsrfcCst}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. )rRrarf)rerrrlockstep_stream_one_way_pair'srgcCsttS)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. )rPrgrrrrlockstep_stream_pairBs rh)r!rZ_highlevel_genericrrabcrrr Finalr2rCrJrOrPrQrRrarfrgrhrrrrs    ?rI!&R^