tjddlZddlZddlmZmZgdZdZdejddfdZdd Z d Z dd Z dd Z Gd dZ dZy)N) is_dunder is_private) StateQueuecreate_state_queue_monitor_task create_task decorate_taskhandle_task_resulttaskSTOPr returnc |jy#tj$rYyt$rt j d|YywxYw)NzException raised by task = %r)resultasyncioCancelledError Exceptionlogging exceptionr s L/DATA/.local/lib/python3.12/site-packages/trame_server/utils/asynchronous.pyr r sDA  ! !  A94@AsA A A cb|tj}t|j|S)a Create a task from a coroutine while also attaching a done callback so any exception or error could be caught and reported. :param coroutine: A coroutine to execute as an independent task :param loop: Optionally provide the loop on which the task should be scheduled on. By default we will use the current running loop. :return: The decorated task :rtype: asyncio.Task )rget_event_loopr r) coroutineloops rrrs- |%%' )))4 55c0|jt|S)a  Decorate a task by attaching a done callback so any exception or error could be caught and reported. :param task: A coroutine to execute as an independent task :type task: asyncio.Task :return: The same task object :rtype: asyncio.Task )add_done_callbackr rs rr r -s -. KrcLKd}|r|jrtj|d{n\|j}t |t r |t k(r3d}n0|j5|jj|ddd|ryy7f#1swYxYww)NTF) emptyrsleep get_nowait isinstancestr QUEUE_EXITstateupdate)serverqueuedelay_monitor_queuemsgs r_queue_update_stater,<s|N  ;;=--& & &""$C#s#*$%*N\\LL'',"  &"\s--B$B=B$-B B$B$B!B$c0tt|||S)a Create and schedule a task to watch over the provided queue to update a server state. This is especially useful when using a multiprocess executor and you want to report progress into your current server. :param server: A coroutine to execute as an independent task :type server: trame_server.core.Server :param queue: A queue instance meant to exchange state from the parallel process to the given server :type queue: multiprocessing.Queue :param delay: Time to sleep in seconds before processing the queue once emptied :type delay: float :return: The monitoring task :rtype: asyncio.Task )r))rr,)r'r(r)s rrrKs* *65F GGrc^eZdZdZddZedZdZdZdZ dZ dZ d Z d Z d Zd Zy )ra Class use to decorate a multiprocessing.Queue inside your external process to simulate your server state object. :param queue: A queue instance meant to exchange state from the parallel process to the given server :type queue: multiprocessing.Queue :param auto_flush: Should you manage the state update phase or just propagate as soon as you update a property :type auto_flush: Boolean cJ||_i|_i|_||_d|_y)Nr)_queue_pending_update _pushed_state _auto_flush _ctx_count)selfr( auto_flushs r__init__zStateQueue.__init__qs( !%rc|jS)z%Provide access to the decorated queue)r0r5s rr(zStateQueue.queuexs{{rcl|jj||jj|SN)r1getr2r5keys r __getitem__zStateQueue.__getitem__}s,##''T-?-?-C-CC-HIIrc\||j|<|jr|jyyr;)r1r3flushr5r>values r __setitem__zStateQueue.__setitem__s)$)S!    JJL rct|rtt|St|r|jj |S|j |Sr;)rgetattrobjectr__dict__r<r?r=s r __getattr__zStateQueue.__getattr__sC S>63' ' c?==$$S) )$$rc^t|r||j|<y|j||yr;)rrHrDrBs r __setattr__zStateQueue.__setattr__s' c?!&DMM#    S% (rct|jj||jr|jyy)z Update the distributed state from a set of key/value pair :param _dict: A dict containing one or many key/value pair :type _dict: dict N)r1r&r3rA)r5_dicts rr&zStateQueue.updates/ ##E*    JJL rct|jrR|jj|j|jj |ji|_yy)z.Explicitly push any local change to the queue.N)lenr1r0 put_nowaitr2r&r9s rrAzStateQueue.flushsO t## $ KK " "4#7#7 8    % %d&:&: ;#%D  %rcB|jjty)z8Release the monitoring task as we are done with our workN)r0rPr$r9s rexitzStateQueue.exits z*rc0|xjdz c_|S)Nr)r4r9s r __enter__zStateQueue.__enter__s 1 rc|xjdzc_|jdk(r!|j|jyy)Nrr)r4rArR)r5exc_type exc_value exc_tracebacks r__exit__zStateQueue.__exit__s3 1 ??a  JJL IIK rN)T)__name__ __module__ __qualname____doc__r7propertyr(r?rDrIrKr&rArRrTrYrrrrcsN J %) &+rrcfd}|S)z.wrappersD$)&)*rr_)rdres` rr r s+ Nrr;)r)rrrr__all__r$Taskr rr r,rrr r_rrris]#  AW\\AdA6$  -H0QQhr