
    j                     n   d dl mZmZ ddlmZmZmZmZ ddlm	Z	 ddl
mZmZ ddlmZmZ ddlmZ dd	lmZ d
ej(                  dedededededej.                  dedej2                  dej4                  fdZdej8                  d
ej(                  dej2                  dedededee   dej4                  fdZy)    )ListOptional   )queue_consumermessagesmessage_queuestreamer)batch_manager_constuctors)attachments_preprocessorbatching_preprocessor) attachments_extraction_processormessage_processors)replay_manager   )base_upload_managerfile_uploaderuse_batchinguse_attachment_extraction#min_base64_embedded_attachment_sizen_consumersmax_queue_sizemessage_processorurl_overridefallback_replay_managerreturnc	           	      ~    t        || |||||      }	t        j                  |	|||      }
|j                  |
       |	S )N)r   r   r   r   r   r   r   )messages_streamermin_attachment_sizer   	is_active)construct_streamerr   AttachmentsExtractionProcessor	add_first)r   r   r   r   r   r   r   r   r   r	   attachment_extractions              /Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/streamer_constructors.pyconstruct_online_streamerr%      sZ     "+# 7!";%H 	)GG& C%/		
   56O    c           	      R   t        j                  |      }t        |      D cg c]  }t        j                  || d|        }	}|rt        j                  |      nd }
t        j                  ||	|t        j                  |
      t        j                  |      |      }|S c c}w )N)
max_lengthQueueConsumerThread_)queuer   name)r*   queue_consumersr   batch_preprocessorr   r   )r   MessageQueueranger   QueueConsumerr
   create_batch_managerr	   Streamerr   BatchingPreprocessorr   AttachmentsPreprocessor)r   r   r   r   r   r   r   message_queue_ir,   batch_manager	streamer_s               r$   r    r    7   s     	""n=  {#; $A 	$$ /'s+	

 $  ;  	"66~F  !!'#0EEmT!9!Q!Q%"
 !8	I 5;s   "B$N)typingr   r    r   r   r   r	   batchingr
   preprocessingr   r   
processorsr   r   replayr   file_uploadr   BaseFileUploadManagerboolintChainedMessageProcessorstrReplayManagerr2   r%   BaseMessageProcessorr     r&   r$   <module>rH      s   !  0 M " - &<<    $  *-	 
     *AA    ,99   F')>>'&<<' ,99' 	'
 '  $' SM' 'r&   