+jddlZddlZddlmZmZgdZdZdejddfdZdd Z d Z dd Z dd Z Gd dZ dZdS)N) is_dunder is_private) StateQueuecreate_state_queue_monitor_task create_task decorate_taskhandle_task_resulttaskSTOPr returnc |dS#tj$rYdSt$rt jd|YdSwxYw)NzException raised by task = %r)resultasyncioCancelledError Exceptionlogging exceptionr s Y/DATA/AppData/hermes/venv/lib/python3.11/site-packages/trame_server/utils/asynchronous.pyr r sqA  !     AAA94@@@@@@AsA A  A cp|tj}t||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 rrrs3 |%'' )))44 5 55c:|t|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 -... KrcXKd}|r|rtj|d{Vnp|}t |t r|t krd}n9|j5|j|dddn #1swxYwY|dSdS)NTF) emptyrsleep get_nowait isinstancestr QUEUE_EXITstateupdate)serverqueuedelay_monitor_queuemsgs r_queue_update_stater,<s N  - ;;== --&& & & & & & & & &""$$C#s## -*$$%*N\--L'',,,---------------  - - - - -s4BB"Bc@tt|||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"* *65FFF G GGrcfeZdZdZddZedZdZdZdZ dZ d Z d Z d Z d Zd ZdS)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 TcL||_i|_i|_||_d|_dS)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 {rch|j||j|SN)r1getr2r5keys r __getitem__zStateQueue.__getitem__}s,#''T-?-C-CC-H-HIIIrcT||j|<|jr|dSdSr;)r1r3flushr5r>values r __setitem__zStateQueue.__setitem__s5$)S!    JJLLLLL  rct|rtt|St|r|j|S||Sr;)rgetattrobjectr__dict__r<r?r=s r __getattr__zStateQueue.__getattr__sX S>> (63'' ' c?? *=$$S)) )$$$rcht|r ||j|<dS|||dSr;)rrHrDrBs r __setattr__zStateQueue.__setattr__s> c?? )!&DM#      S% ( ( ( ( (rct|j||jr|dSdS)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.updatesA ##E***    JJLLLLL  rct|jrG|j|j|j|ji|_dSdS)z.Explicitly push any local change to the queue.N)lenr1r0 put_nowaitr2r&r9s rrAzStateQueue.flushs` t# $ $ & K " "4#7 8 8 8   % %d&: ; ; ;#%D  & &rcD|jtdS)z8Release the monitoring task as we are done with our workN)r0rPr$r9s rexitzStateQueue.exits z*****rc&|xjdz c_|S)Nr)r4r9s r __enter__zStateQueue.__enter__s 1 rc|xjdzc_|jdkr*||dSdS)Nrr)r4rArR)r5exc_type exc_value exc_tracebacks r__exit__zStateQueue.__exit__sE 1 ?a   JJLLL IIKKKKK rN)T)__name__ __module__ __qualname____doc__r7propertyr(r?rDrIrKr&rArRrTrYrrrrcs  XJJJ %%%)))   &&&+++rrcfd}|S)z.wrappers&DD$)&))*****rr_)rdres` rr r s#+++++ Nrr;)r)rrrr__all__r$Taskr rr r,rrr r_rrris ########    AW\AdAAAA6666$    - - - -HHHH0QQQQQQQQhr