o
    ùT·jÝ  ã                   @  sN   d dl mZ d dlZd dlZd dlmZ d dlZddlm	Z	 G dd„ dƒZ
dS )é    )ÚannotationsN)Údequeé   )ÚWebSocketQueueFullErrorc                   @  sl   e Zd ZdZd$d%dd„Zd&dd„Zd'dd„Zd(dd„Zd)dd„Zd*dd„Z	d+dd„Z
d,dd„Zd-d!d"„Zd#S ).Ú	SendQueuezèBounded byte-size queue for outgoing WebSocket messages.

    Messages are stored as pre-serialized strings. The queue enforces a
    maximum byte budget so that unbounded buffering cannot occur during
    reconnection windows.
    é   Ú	max_bytesÚintÚreturnÚNonec                 C  s&   g | _ d| _|| _t ¡ | _d | _d S ©Nr   )Ú_queueÚ_bytesÚ
_max_bytesÚ	threadingÚLockÚ_lockÚ_flush_done)Úselfr   © r   úU/home/dinkstrade/pdmp-scanner/venv/lib/python3.10/site-packages/openai/_send_queue.pyÚ__init__   s
   

zSendQueue.__init__ÚdataÚstrc                 C  sp   t | d¡ƒ}| j�$ | j| | jkrtdƒ‚| j ||f¡ |  j|7  _W d  ƒ dS 1 s1w   Y  dS )zŽAppend *data* to the queue.

        Raises :class:`WebSocketQueueFullError` if the message would
        exceed the byte-size limit.
        zutf-8z%send queue is full, message discardedN)ÚlenÚencoder   r   r   r   r   Úappend)r   r   Úbyte_lengthr   r   r   Úenqueue   s   "üzSendQueue.enqueueÚsendútyping.Callable[[str], object]c              	   C  s¶   t |  ¡  }tjƒr| ¡  t |  ¡  }tjƒs
z<|rM|d \}}||ƒ | j� | ¡  |  j|8  _W d  ƒ n1 s>w   Y  |sW |  |¡ dS W |  |¡ dS |  |¡ w )z«Send every queued message via *send*.

        If *send* raises, the failing message and all subsequent messages
        are re-queued and the error is re-raised.
        r   N)	Ú
isinstanceÚ_begin_flushr   ÚEventÚwaitr   Úpopleftr   Ú
_end_flush©r   r   Úpendingr   r   r   r   r   Ú
flush_sync(   s   ÿþýùzSendQueue.flush_syncú0typing.Callable[[str], typing.Awaitable[object]]c              	   Ã  sÎ   �t |  ¡  }tjƒr!tjj|jdd�I dH  t |  ¡  }tjƒsz?|rY|d \}}||ƒI dH  | j� | 	¡  |  j
|8  _
W d  ƒ n1 sJw   Y  |s$W |  |¡ dS W |  |¡ dS |  |¡ w )z$Async variant of :meth:`flush_sync`.T)Úabandon_on_cancelNr   )r!   r"   r   r#   ÚanyioÚ	to_threadÚrun_syncr$   r   r%   r   r&   r'   r   r   r   Úflush_async;   s    €ýþýùzSendQueue.flush_asyncú(deque[tuple[str, int]] | threading.Eventc                 C  sj   | j �( | jd ur| jW  d   ƒ S t| jƒ}| j ¡  t ¡ | _|W  d   ƒ S 1 s.w   Y  d S ©N)r   r   r   r   Úclearr   r#   ©r   r(   r   r   r   r"   L   s   
þ


$ùzSendQueue._begin_flushr(   údeque[tuple[str, int]]c                 C  sZ   | j �  t|ƒ| j | _| jd usJ ‚| j ¡  d | _W d   ƒ d S 1 s&w   Y  d S r1   )r   Úlistr   r   Úsetr3   r   r   r   r&   V   s   
"üzSendQueue._end_flushú	list[str]c                 C  sf   | j �& dd„ | jD ƒ}|  jtdd„ | jD ƒƒ8  _| j ¡  |W  d  ƒ S 1 s,w   Y  dS )z&Remove and return all queued messages.c                 S  s   g | ]\}}|‘qS r   r   )Ú.0r   Ú_r   r   r   Ú
<listcomp>`   s    z#SendQueue.drain.<locals>.<listcomp>c                 s  s   � | ]\}}|V  qd S r1   r   )r8   r9   r   r   r   r   Ú	<genexpr>a   s   € z"SendQueue.drain.<locals>.<genexpr>N)r   r   r   Úsumr2   )r   Úitemsr   r   r   Údrain]   s   
$üzSendQueue.drainc                 C  s4   | j � t| jƒW  d   ƒ S 1 sw   Y  d S r1   ©r   r   r   ©r   r   r   r   Ú__len__e   s   $ÿzSendQueue.__len__Úboolc                 C  s8   | j � t| jƒdkW  d   ƒ S 1 sw   Y  d S r   r?   r@   r   r   r   Ú__bool__i   s   $ÿzSendQueue.__bool__N)r   )r   r	   r
   r   )r   r   r
   r   )r   r    r
   r   )r   r*   r
   r   )r
   r0   )r(   r4   r
   r   )r
   r7   )r
   r	   )r
   rB   )Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r   r)   r/   r"   r&   r>   rA   rC   r   r   r   r   r      s    







r   )Ú
__future__r   Útypingr   Úcollectionsr   Úanyio.to_threadr,   Ú_exceptionsr   r   r   r   r   r   Ú<module>   s    