File: //usr/local/lib/python3.10/dist-packages/anyio/streams/__pycache__/memory.cpython-310.pyc
o
���g6) � @ sT d dl mZ d dlZd dlmZmZ d dlmZmZ d dl m
Z
d dl m Z m
Z
mZ ddlmZmZmZmZ dd lmZmZ dd
lmZmZmZ dd lm Z ed �Z ed
dd�Z eddd�Z G dd� de
�Z edd�G dd� de e ��Z!edd�G dd� de e ��Z"edd�G dd� de e ee ��Z#edd�G d d � d e e ee ��Z$dS ) � )� annotationsN)� OrderedDict�deque)� dataclass�field)�
TracebackType)�Generic�
NamedTuple�TypeVar� )�BrokenResourceError�ClosedResourceError� EndOfStream�
WouldBlock)�TaskInfo�get_current_task)�Event�ObjectReceiveStream�ObjectSendStream)�
checkpoint�T_Item�T_coT)� covariant�T_contra)�
contravariantc @ s> e Zd ZU ded< ded< ded< ded< ded< ded< d S )
� MemoryObjectStreamStatistics�int�current_buffer_used�float�max_buffer_size�open_send_streams�open_receive_streams�tasks_waiting_send�tasks_waiting_receiveN)�__name__�
__module__� __qualname__�__annotations__� r( r( �?/usr/local/lib/python3.10/dist-packages/anyio/streams/memory.pyr s
r F)�eqc @ s> e Zd ZU eded�Zded< edd�Zded< d
d
d �Zd S )�MemoryObjectItemReceiverF��init�default_factoryr � task_info)r- r �item�return�strc C s( t | dd �}| jj� d| j� d|�d�S )Nr0 z (task_info=z, item=�))�getattr� __class__r$ r/ )�selfr0 r( r( r) �__repr__) s z!MemoryObjectItemReceiver.__repr__N)r1 r2 ) r$ r% r& r r r/ r' r0 r7 r( r( r( r) r+ $ s
r+ c @ s� e Zd ZU e� Zded< eded�Zded< eddd�Zd ed
< eddd�Z d ed < ede
d�Z d ed
< ede
d�Z ded< ddd�Z
dS )�MemoryObjectStreamStater r Fr, z
deque[T_Item]�bufferr �r- �defaultr �open_send_channels�open_receive_channelsz4OrderedDict[Event, MemoryObjectItemReceiver[T_Item]]�waiting_receiverszOrderedDict[Event, T_Item]�waiting_sendersr1 r c C s* t t| j�| j| j| jt| j�t| j��S �N)r �lenr9 r r< r= r? r> �r6 r( r( r) �
statistics= s �z"MemoryObjectStreamState.statisticsN�r1 r )r$ r% r& r r r' r r9 r< r= r r> r? rC r( r( r( r) r8 0 s
��r8 c @ s� e Zd ZU ded< eddd�Zded< d'd d
�Zd(d d
�Zd(dd�Zd)dd�Z d'dd�Z
d'dd�Z d*dd�Z d)dd�Z
d+d"d#�Zd'd$d%�Zd&S ),�MemoryObjectReceiveStreamz MemoryObjectStreamState[T_co]�_stateFr: �bool�_closedr1 �Nonec C � | j jd7 _d S �N� )rF r= rB r( r( r) �
__post_init__M � z'MemoryObjectReceiveStream.__post_init__r c C s` | j rt�| jjr | jjjdd�\}}| jj�|� |�� | jjr(| jj�� S | jj s.t
�t �)a�
Receive the next item if it can be done without waiting.
:return: the received item
:raises ~anyio.ClosedResourceError: if this send stream has been closed
:raises ~anyio.EndOfStream: if the buffer is empty and this stream has been
closed from the sending end
:raises ~anyio.WouldBlock: if there are no items in the buffer and no tasks
waiting to send
F��last) rH r
rF r? �popitemr9 �append�set�popleftr< r r )r6 �
send_eventr0 r( r( r) �receive_nowaitP s z(MemoryObjectReceiveStream.receive_nowaitc
� s� �t � I d H z| �� W S tyL t� }tt � }|| jj|< z|�� I d H W | jj� |d � n | jj� |d � w z|j
W Y S t yK t �w w r@ )
r rV r r r+ r rF r> �wait�popr0 �AttributeErrorr )r6 �
receive_event�receiverr( r( r) �receivel s" �
$ ��z!MemoryObjectReceiveStream.receive� MemoryObjectReceiveStream[T_co]c C � | j rt�t| jd�S )a
Create a clone of this receive stream.
Each clone can be closed separately. Only when all clones have been closed will
the receiving end of the memory stream be considered closed by the sending ends.
:return: the cloned stream
�rF )rH r
rE rF rB r( r( r) �clone� �
z MemoryObjectReceiveStream.clonec C sV | j s%d| _ | j jd8 _| jjdkr't| jj�� �}|D ]
}|�� q dS dS dS �z�
Close the stream.
This works the exact same way as :meth:`aclose`, but is provided as a special
case for the benefit of synchronous callbacks.
TrL r N)rH rF r= �listr? �keysrS )r6 � send_events�eventr( r( r) �close� s
�z MemoryObjectReceiveStream.closec � � �| � � d S r@ �rg rB r( r( r) �aclose� � � z MemoryObjectReceiveStream.acloser c C �
| j �� S �zj
Return statistics about the current state of this stream.
.. versionadded:: 3.0
�rF rC rB r( r( r) rC � �
z$MemoryObjectReceiveStream.statisticsc C � | S r@ r( rB r( r( r) � __enter__� � z#MemoryObjectReceiveStream.__enter__�exc_type�type[BaseException] | None�exc_val�BaseException | None�exc_tb�TracebackType | Nonec C � | � � d S r@ ri �r6 rs ru rw r( r( r) �__exit__� � z"MemoryObjectReceiveStream.__exit__c C �6 | j stjd| jj� dt| �d�d�t| d� d S d S �Nz
Unclosed <z at �x�>)�source�rH �warnings�warnr5 r$ �id�ResourceWarningrB r( r( r) �__del__� �
��z!MemoryObjectReceiveStream.__del__N�r1 rI )r1 r )r1 r] rD �rs rt ru rv rw rx r1 rI )r$ r% r& r' r rH rM rV r\ r` rg rj rC rq r{ r� r( r( r( r) rE H s
rE c @ s� e Zd ZU ded< eddd�Zded< d(d d
�Zd)d
d�Zd)dd�Zd*dd�Z d(dd�Z
d(dd�Z d+dd�Z d*dd �Z
d,d#d$�Zd(d%d&�Zd'S )-�MemoryObjectSendStreamz!MemoryObjectStreamState[T_contra]rF Fr: rG rH r1 rI c C rJ rK )rF r< rB r( r( r) rM � rN z$MemoryObjectSendStream.__post_init__r0 r c C s� | j rt�| jjs t�| jjr+| jjjdd�\}}|j�� s'||_ |�
� dS | jjst | jj �| jj
k r>| jj �|� dS t�)a�
Send an item immediately if it can be done without waiting.
:param item: the item to send
:raises ~anyio.ClosedResourceError: if this send stream has been closed
:raises ~anyio.BrokenResourceError: if the stream has been closed from the
receiving end
:raises ~anyio.WouldBlock: if the buffer is full and there are no tasks waiting
to receive
FrO N)rH r
rF r= r r> rQ r/ �has_pending_cancellationr0 rS rA r9 r rR r )r6 r0 rZ r[ r( r( r) � send_nowait� s
�z"MemoryObjectSendStream.send_nowaitc � s� �t � I dH z| �|� W dS tyJ t� }|| jj|< z |�� I dH W n ty8 | jj�|d� � w || jjv rG| jj|= t d�Y dS w )a�
Send an item to the stream.
If the buffer is full, this method blocks until there is again room in the
buffer or the item can be sent directly to a receiver.
:param item: the item to send
:raises ~anyio.ClosedResourceError: if this send stream has been closed
:raises ~anyio.BrokenResourceError: if the stream has been closed from the
receiving end
N)
r r� r r rF r? rW �
BaseExceptionrX r )r6 r0 rU r( r( r) �send� s$ �
�
��zMemoryObjectSendStream.send� MemoryObjectSendStream[T_contra]c C r^ )a
Create a clone of this send stream.
Each clone can be closed separately. Only when all clones have been closed will
the sending end of the memory stream be considered closed by the receiving ends.
:return: the cloned stream
r_ )rH r
r� rF rB r( r( r) r` ra z MemoryObjectSendStream.clonec C sb | j s+d| _ | j jd8 _| jjdkr-t| jj�� �}| jj�� |D ]
}|�� q$dS dS dS rb )rH rF r< rc r> rd �clearrS )r6 �receive_eventsrf r( r( r) rg s
�z MemoryObjectSendStream.closec � rh r@ ri rB r( r( r) rj ! rk z MemoryObjectSendStream.acloser c C rl rm rn rB r( r( r) rC $ ro z!MemoryObjectSendStream.statisticsc C rp r@ r( rB r( r( r) rq , rr z MemoryObjectSendStream.__enter__rs rt ru rv rw rx c C ry r@ ri rz r( r( r) r{ / r| z MemoryObjectSendStream.__exit__c C r} r~ r� rB r( r( r) r� 7 r� z MemoryObjectSendStream.__del__Nr� )r0 r r1 rI )r1 r� rD r� )r$ r% r& r' r rH rM r� r� r` rg rj rC rq r{ r� r( r( r( r) r� � s
r� )%�
__future__r r� � collectionsr r � dataclassesr r �typesr �typingr r r
� r r
r r �_core._testingr r �abcr r r �lowlevelr r r r r r+ r8 rE r� r( r( r( r) �<module> s, u