
    j	                     t    d dl Z d dlmZ d dlZd dlmZmZmZ  ed      Z G d dee         Z	de
de
d	e
fd
Zy)    N)Empty)TypeVarOptionalGenericTc                       e Zd ZdZddee   fdZdeddfdZdeddfdZ	d	e
defd
ZdefdZdefdZdefdZdefdZy)MessageQueuea1  A thread-safe message queue implementation using a double-ended queue (deque).

    This queue allows putting messages at either end and retrieving them in a FIFO manner.
    It supports optional maximum length limiting and blocking retrieval with timeout.

    Once a bounded length queue is full, when new items are added, a corresponding number
    of items are discarded from the opposite end.

    Args:
        max_length (Optional[int]): Maximum number of messages the queue can hold.
            If None, the queue size is unlimited.

    The queue implements the following operations:
    - put(): Add a message to the front of the queue
    - put_back(): Add a message to the back of the queue
    - get(): Remove and return message from the back of the queue
    - empty(): Check if the queue is empty
    N
max_lengthc                 z    t        j                  |      | _        t        j                         | _        || _        y )N)maxlen)collectionsdeque_deque	threading	Condition
_not_emptymax_size)selfr
   s     ~/Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/message_queue.py__init__zMessageQueue.__init__   s+    ,7,=,=Z,P#--/"    messagereturnc                     | j                   5  | j                  j                  |       | j                   j                          d d d        y # 1 sw Y   y xY wN)r   r   
appendleftnotifyr   r   s     r   putzMessageQueue.put"   s5    __KK""7+OO""$ __   6AAc                     | j                   5  | j                  j                  |       | j                   j                          d d d        y # 1 sw Y   y xY wr   )r   r   appendr   r   s     r   put_backzMessageQueue.put_back'   s5    __KKw'OO""$ __r    timeoutc                      j                   5  ||dk  rt        d       j                   j                   fd|       t         j                        dk(  rt
         j                  j                         cd d d        S # 1 sw Y   y xY w)Nr   z''timeout' must be a non-negative numberc                  4    t         j                        dkD  S Nr   lenr   r   s   r   <lambda>z"MessageQueue.get.<locals>.<lambda>1   s    S-=-Ar   )r$   )r   
ValueErrorwait_forr)   r   r   pop)r   r$   s   ` r   getzMessageQueue.get,   sg    __'A+ !JKKOO$$%A7$S4;;1$;;??$ __s   A*BBc                 2    t        | j                        dk(  S r'   r(   r*   s    r   emptyzMessageQueue.empty8   s    4;;1$$r   c                     | j                   y| j                  5  t        | j                        | j                   k  cd d d        S # 1 sw Y   y xY w)NT)r   r   r)   r   r*   s    r   accept_put_without_discardingz*MessageQueue.accept_put_without_discarding;   s5    == __t{{#dmm3 __s   "AAc                 ,    t        | j                        S r   r(   r*   s    r   sizezMessageQueue.sizeA       4;;r   c                 ,    t        | j                        S r   r(   r*   s    r   __len__zMessageQueue.__len__D   r6   r   r   )__name__
__module____qualname____doc__r   intr   r   r   r#   floatr/   boolr1   r3   r5   r8    r   r   r	   r	   	   s    &#8C= #
%1 % %
% %d %

%5 
%Q 
%%t %4t 4 c     r   r	   maximal_queue_sizebatch_factorr   c                 ,    |dkD  rt        | |z        S | S r'   )r=   )rA   rB   s     r   calculate_max_queue_sizerD   H   s#     a%455!!r   )r   queuer   r   typingr   r   r   r   r	   r=   rD   r@   r   r   <module>rG      sM       - -CL< 71: < ~""" 	"r   