
    j                     T    d dl Z d dlmZmZ ddlmZ ddlmZ ddlmZ  G d d	      Zy)
    N)TypeDict   )messages   )base_batcher)flushing_threadc                       e Zd Zdeeej                     ej                  f   ddfdZ	ddZ
ddZdej                  defdZdej                  ddfd	Zdefd
ZddZy)BatchManagermessage_to_batcher_mappingreturnNc                     || _         t        j                  t        | j                   j	                                     | _        t        j                         | _        y )N)batchers)	_message_to_batcher_mappingr	   FlushingThreadlistvalues_flushing_thread	threadingRLock_lock)selfr   s     /Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/batching/batch_manager.py__init__zBatchManager.__init__	   sG     ,F( / > >$::AACD!
 __&
    c                 8    | j                   j                          y N)r   startr   s    r   r   zBatchManager.start   s    ##%r   c                     | j                   5  | j                  j                          | j                          d d d        y # 1 sw Y   y xY wr   )r   r   closeflushr   s    r   stopzBatchManager.stop   s.    ZZ!!'')JJL	 ZZs   +AA
messagec                 h    |yt        |d      r|j                  S |j                  | j                  v S )NFsupports_batching)hasattrr&   	__class__r   r   r$   s     r   message_supports_batchingz&BatchManager.message_supports_batching   s8    ?7/0,,,  D$D$DDDr   c                     | j                   5  | j                  t        |         j                  |       d d d        y # 1 sw Y   y xY wr   )r   r   typeaddr)   s     r   process_messagezBatchManager.process_message(   s0    ZZ,,T'];??H ZZs	   (>Ac                     | j                   5  t        | j                  j                         D cg c]  }|j	                          c}      cd d d        S c c}w # 1 sw Y   y xY wr   )r   allr   r   is_emptyr   batchers     r   r1   zBatchManager.is_empty,   sZ    ZZ $(#C#C#J#J#L#L $$&#L Z Zs   "AA
AAA$c                     | j                   5  | j                  j                         D ]  }|j                           	 d d d        y # 1 sw Y   y xY wr   )r   r   r   r"   r2   s     r   r"   zBatchManager.flush5   s6    ZZ;;BBD E ZZs   0AA)r   N)__name__
__module____qualname__r   r   r   BaseMessager   BaseBatcherr   r   r#   boolr*   r.   r1   r"    r   r   r   r      s    
'$(%%&(@(@@%

'
 

'&E1E1E E$ EIx';'; I I$  r   r   )	r   typingr   r    r   r   r	   r   r;   r   r   <module>r>      s        0  0 r   