
    j7                         d dl Z d dlZd dlZd dlmZmZ d dlmZ ddlm	Z	m
Z
mZ ddlmZ ddlmZ dd	lmZmZ dd
lmZ  ej&                  e      Z G d d      Zy)    N)ListOptional)base_upload_manager   )messagesmessage_queuequeue_consumer   )_logging)synchronization)attachments_preprocessorbatching_preprocessor)replay_managerc                   h   e Zd Zdej                  ej                     deej                     de
j                  dej                  dej                  dej"                  ddfd	Zd
ej                  ddfdZdee   defdZddee   dedefdZdefdZdee   defdZdefdZdefdZddZddZy)Streamerqueuequeue_consumersr   batch_preprocessorfile_uploaderfallback_replay_managerreturnNc                 x   t        j                         | _        || _        || _        || _        || _        || _        || _        d| _	        d| _
        | j                          | j                  j                          | j                  j                  | j                         | j                  j                          y )NFT)	threadingRLock_lock_message_queue_queue_consumers_attachments_preprocessor_batch_preprocessor_file_upload_manager_fallback_replay_manager_drain_idle_start_queue_consumersstartset_replay_callbackput)selfr   r   r   r   r   r   s          y/Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/streamer.py__init__zStreamer.__init__   s     __&
# /)A&#5 $1!(?%
##%  &&(%%99$((C%%++-    messagec                    | j                   5  | j                  r
	 d d d        y d| _        	 | j                  j	                  |      }| j
                  j	                  |      }|a| j                  j                         du r*t        j                  t        j                  dt               | j                  j                  |       d| _        d d d        y # t        $ r"}t        j                  d||       Y d }~6d }~ww xY w# 1 sw Y   y xY w)NFzThe message queue size limit has been reached. The new message has been added to the queue, and the oldest message has been discarded.)loggerz)Failed to process message by streamer: %s)exc_infoT)r   r"   r#   r   
preprocessr   r   accept_put_without_discardingr   log_once_at_levelloggingWARNINGLOGGERr'   	Exceptionerror)r(   r,   preprocessed_messageexs       r)   r'   zStreamer.put0   s    ZZ{{ Z DJ'+'E'E'P'P($
 (,'?'?'J'J(($
 (3**HHJeS 22#OO e#)
 ''++,@A
 DJ; Z2  ?b   3 Zs:   DDBCD	D C;6D;D  DDtimeoutc                      j                   5  t        j                   fd|d       d _        ddd        j                  j                           j                  j                           j                  j                  |        j                  |        j                           j                  j                         S # 1 sw Y   xY w)z/
        Stops data processing threads
        c                       j                   S Nr#   r(   s   r)   <lambda>z Streamer.close.<locals>.<lambda>V   	    tzzr+   皙?check_functionr:   
sleep_timeTN)r   r   wait_for_doner"   r   stopr!   closejoinflush_close_queue_consumersr   emptyr(   r:   s   ` r)   rH   zStreamer.closeP   s     ZZ))1
 DK  	  %%'%%++-%%**73

7##%""((** Zs   #CCupload_sleep_timec                 N     j                   5  t        j                   fd|d       d d d         j                  j	                           j
                  j                  r j
                  j	                          t        j                         }t        j                   fd|d       t        j                         |z
  }|||z
  }|dk  rd} j                  j	                  ||      }|xr  j                         }t        j                  d|        |S # 1 sw Y   xY w)	Nc                       j                   S r=   r>   r?   s   r)   r@   z Streamer.flush.<locals>.<lambda>k   rA   r+   rB   rC   c                  $     j                         S r=   )	_all_doner?   s   r)   r@   z Streamer.flush.<locals>.<lambda>y   s    4>>#3r+   g        g      ?)r:   rE   zStreamer flushed completely: )r   r   rF   r   rJ   r!   has_server_connectiontimer    rR   r5   debug)r(   r:   rN   
start_timeelapsed_timeupload_flushedflusheds   `      r)   rJ   zStreamer.flushe   s    ZZ))1  	  &&(((>>))//1YY[
%%3	
 yy{Z/,G} 2288(9 9 
 !5T^^%54WI>?G Zs   DD$c                     | j                         xr6 | j                  j                         xr | j                  j	                         S r=   )workers_idlingr   rL   r   is_emptyr?   s    r)   rR   zStreamer._all_done   sA    ! 4##))+4((113	
r+   c                 :    | j                   j                  |      S )zXReturns the number of failed file uploads. Blocking - waits for all uploads to complete.)r:   )r    failed_uploadsrM   s     r)    __internal_api__failed_uploads__z)Streamer.__internal_api__failed_uploads__   s    ((777HHr+   c                 f    t        | j                  D cg c]  }|j                   c}      S c c}w r=   )allr   idlingr(   consumers     r)   r[   zStreamer.workers_idling   s+    D4I4IJ4IHOO4IJKKJs   .c                 6    | j                   j                         S r=   )r   sizer?   s    r)   
queue_sizezStreamer.queue_size   s    ""''))r+   c                 F    | j                   D ]  }|j                           y r=   )r   r%   rc   s     r)   r$   zStreamer._start_queue_consumers       --HNN .r+   c                 F    | j                   D ]  }|j                           y r=   )r   rH   rc   s     r)   rK   zStreamer._close_queue_consumers   ri   r+   )   )r   N) __name__
__module____qualname__r   MessageQueuer   BaseMessager   r	   QueueConsumerr   AttachmentsPreprocessorr   BatchingPreprocessorr   BaseFileUploadManagerr   ReplayManagerr*   r'   r   intboolrH   floatrJ   rR   r_   r[   rg   r$   rK    r+   r)   r   r      s   .))(*>*>?. n::;. #;"R"R	.
 2FF. +@@. "0!=!=. 
.68// D @+Xc] +t +*'Xe_ ' 'T 'R
4 
I IC IL L*C *r+   r   )r   r3   rT   typingr   r   opik.file_uploadr    r   r   r	   r   r   preprocessingr   r   replayr   	getLoggerrl   r5   r   ry   r+   r)   <module>r      sJ       ! 0 5 5   # 
		8	$Q Qr+   