o NTh@sddlmZddlZddlmZddlmZddlmZddl Z ddl m Z m Z erBddl mZmZmZmZdd lmZdd lmZedddZGdddZdddZed ddZdS)!) annotationsN) OrderedDict)contextmanager) TYPE_CHECKING)ConflictDetectoris_main_thread) AsyncIteratorCallable GeneratorIterable) FrameType)Selfsignals Iterable[int]handlerHCallable[[int, FrameType | None], object] | int | signal.Handlers | NonereturnGenerator[None, None, None]c cspi}z$t|D] }t||||<qdVW|D] \}}t||qdS|D] \}}t||q,wN)setsignalitems)rroriginal_handlerssignumoriginal_handlerrN/home/air/segue/gemini/back/venv/lib/python3.10/site-packages/trio/_signals.py_signal_handler6s rc@s>eZdZdddZdddZdd d Zdd d ZdddZdS)SignalReceiverrNonecCs(t|_tj|_td|_d|_dS)Nz8only one task can iterate on a signal receiver at a timeF) r_pendingtriolowlevel ParkingLot_lotr_conflict_detector_closedselfrrr__init__Fs   zSignalReceiver.__init__rintcCs,|jr t|dSd|j|<|jdSr)r'r raise_signalr!r%unpark)r)rrrr_addOs zSignalReceiver._addcs d_dfdd dS)NTrr cs<jrjjdd\}}z t|WdSwdS)NFlast)r!popitemrr,)r_ deliver_nextr)rrr4^s  z9SignalReceiver._redeliver_remaining..deliver_nextrr )r'r(rr3r_redeliver_remainingVs z#SignalReceiver._redeliver_remainingrcCs|Srrr(rrr __aiter__hszSignalReceiver.__aiter__csx|jrtd|j'|js|jIdHntjIdH|jj dd\}}|WdS1s5wYdS)Nz)open_signal_receiver block already exitedFr/) r' RuntimeErrorr&r!r%parkr"r# checkpointr1)r)rr2rrr __anext__ks$zSignalReceiver.__anext__Nr5)rr+rr )rr)rr+)__name__ __module__ __qualname__r*r.r6r7r;rrrrrEs    rrecAsyncIterator[int]r+cCst|tsJt|jS)z/Helper for tests, not public or otherwise used.) isinstancerlenr!)r?rrrget_pending_signal_countzs rCsignal.Signals | int)Generator[AsyncIterator[int], None, None]c's|stdtstdtjtd fd d }z&t|| VWd n1s2wYWd SWd Sw) aA context manager for catching signals. Entering this context manager starts listening for the given signals and returns an async iterator; exiting the context manager stops listening. The async iterator blocks until a signal arrives, and then yields it. Note that if you leave the ``with`` block while the iterator has unextracted signals still pending inside it, then they will be re-delivered using Python's regular signal handling logic. This avoids a race condition when signals arrives just before we exit the ``with`` block. Args: signals: the signals to listen for. Raises: TypeError: if no signals were provided. RuntimeError: if you try to use this anywhere except Python's main thread. (This is a Python limitation.) Example: A common convention for Unix daemons is that they should reload their configuration when they receive a ``SIGHUP``. Here's a sketch of what that might look like using :func:`open_signal_receiver`:: with trio.open_signal_receiver(signal.SIGHUP) as signal_aiter: async for signum in signal_aiter: assert signum == signal.SIGHUP reload_configuration() zNo signals were providedz]Sorry, open_signal_receiver is only possible when running in Python interpreter's main threadrr+frameFrameType | Nonerr csjj|dddS)NT) idempotent) run_sync_soonr.)rrFqueuetokenrrrsz%open_signal_receiver..handlerN)rr+rFrGrr ) TypeErrorrr8r"r#current_trio_tokenrrr6)rrrrJropen_signal_receivers"&   rO)rrrrrr)r?r@rr+)rrDrrE) __future__rr collectionsr contextlibrtypingrr"_utilrrcollections.abcr r r r typesr typing_extensionsrrrrCrOrrrrs"      &  5