jddlZddlZddlZddlmZddlmZddlmZm Z ddlm Z eje Z deed dfd ZGd d ej Zy) N)Optional)connection_monitor) db_managertypes)messages message_idreturnc| tdy)NzMessage ID expected) ValueError)r s /Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/replay/replay_manager.py_check_message_idrs.//c~eZdZdZdej dededeffd Zdfd Z dd Z e de jfd Ze defd Ze j"j$fd ej(de j"ddfdZdeddfdZ ddedeeddfdZdej6ddfdZddZddZddZddZ ddZ!xZ"S) ReplayManagera Manages replaying messages to the server for a connection management system. The ReplayManager is responsible for ensuring that messages which fail to be delivered due to dropped connections are replayed to the server when the connection is restored. It continuously monitors the connection status and triggers failed message replays as necessary. The class runs in its own thread, leveraging the threading.Thread base class. monitor batch_sizebatch_replay_delaytick_interval_secondscHt|ddtj|_||_d|_||_tj|jz|_ d|_ tj|_ tj|||_y)a Initializes the ReplayManager instance. Creates the replay manager thread with a specified connection monitor and tick interval. Args: monitor: An instance of OpikConnectionMonitor used for monitoring the connection. batch_size: The size of batches for processing. batch_replay_delay: The delay (in seconds) between replaying batches of messages. tick_interval_seconds: Interval in seconds between execution ticks. Default is 0.3. Tr)daemonnameNr)rr)super__init__ threadingEvent _stop_running_monitor_replay_callback_tick_interval_secondstime_next_tick_time_next_message_idRLock_message_id_lockr DBManager _db_manager)selfrrrr __class__s rrzReplayManager.__init__s$ ?;&__. @D&;##yy{T-H-HH ! ) 1%//!1 rr NcB|jt| yN)_check_replay_callbackrstart)r)r*s rr.zReplayManager.start@s ##%  rc |jjs+|j|jjs+|jj y#|jj wxYwr,)ris_set_loopr(closer)s rrunzReplayManager.runDs] %((//1 ((//1    " " $D   " " $s AA""A>c|jSr,)r(r3s rdatabase_managerzReplayManager.database_managerLsrc.|jjS)z2Checks if SDK has a connection to the OPIK server.)rhas_server_connectionr3s rr8z#ReplayManager.has_server_connectionPs}}222rmessagestatuscB|j5|j&|j|_|xjdz c_ddd |jj ||y#1swY(xYw#t $r"}t jd|dYd}~yd}~wwxYw)z=Registers a message to be replayed if the connection is lost.Nrr:z1Failed to register message for replay, reason: %sTexc_info)r&r r$r(register_message ExceptionLOGGERerror)r)r9r:exs rr?zReplayManager.register_messageUs  " "!!)%)%:%:"%%*% #      - -gf - E# "  LLCRRV    s#3A' A3'A03 B<BBr ct| |jj|tjj y#t $r"}tjd|dYd}~yd}~wwxYw)zDUnregisters a message from being replayed if the connection is lost.r<z5Failed to un-register message from replay, reason: %sTr=N) rr(update_messager MessageStatus deliveredr@rArB)r)r rCs runregister_messagez ReplayManager.unregister_messagehsj*%     + +:#;#;#E#E ,   LLG    s5A A. A))A.failure_reasonct| |jj||jj |t j jy#t$r"}tjd|dYd}~yd}~wwxYw)zMNotifies the manager that a message was not sent due to a connection failure.)rIr<zCFailed to mark message as send-failed in replay manager, reason: %sTr=N) rrconnection_failedr(rErrFfailedr@rArB)r)r rIrCs rmessage_sent_failedz!ReplayManager.message_sent_failedvs *%  MM + +> + J    + +:#;#;#B#B ,   LLU    sAA B (BB callbackc||_y)z?Sets the callback to be invoked when replaying failed messages.N)r )r)rNs rset_replay_callbackz!ReplayManager.set_replay_callbacks (rc8|jjy)zStop the replay manager.N)rsetr3s rr2zReplayManager.closes  rcD|j|jy)z2Force replay of all failed messages to the server.N)r-_replay_failed_messagesr3s rflushzReplayManager.flushs ##% $$&rc|jtjz }|dkDr|jj| |jj }|t jjk(r*|j|jjtj|jz|_y#t$r"}tjd||Yd}~Md}~wwxYw)NrzXFailed to tick the connection monitor or replay failed messages, will repeat. Reason: %sr=)r#r"rwaitrtickrConnectionStatusconnection_restoredrTresetr@rAwarningr!)r) sleep_timer:rCs rr1zReplayManager._loops))DIIK7 >    # #J / ]]'')F+<<PPP,,. ##% $yy{T-H-HH  NNj    sA!C C6C11C6c2|j tdy)Nz>Replay callback must be set before starting the replay manager)r r r3s rr-z$ReplayManager._check_replay_callbacks#  (P  )rc|j |jj|j}|dkDrtj d|yy#t $r"}tjd|dYd}~yd}~wwxYw)NrzIReplayed %d message(s) that were not sent due to server connection issuesz,Failed to replay failed messages, reason: %sTr=)r-r(replay_failed_messagesr rAinfor@rB)r)replayedrCs rrTz%ReplayManager._replay_failed_messagess ##% ''>>%%H!| _   LL>T    sAA A?A::A?)r Nr,)#__name__ __module__ __qualname____doc__rOpikConnectionMonitorintfloatrr.r4propertyrr'r6boolr8rF registeredr BaseMessager?rHrstrrMrReplayCallbackrPr2rUr1r-rT __classcell__)r*s@rrrs, #99  "  %  B% *"6"6  3t33,6+C+C+N+N%%((  & S T @D/7} ")E,@,@)T)!' I, rr)loggingrr"typingropik.healthcheckrrrr getLoggerrcrArhrThreadrrrrxsY /   8 $0(3-0D0 mI$$mr