o 8e@sddlmZddlZddlmZddlmZmZmZm Z ddl m Z ddl m Z ddlmZddlZdd lmZmZmZerDdd lmZe dddZGdddZdddZe d ddZdS)!) annotationsN) OrderedDict) AsyncIteratorCallable GeneratorIterable)contextmanager) FrameType) TYPE_CHECKING)ConflictDetectoris_main_thread signal_raise)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)rrZoriginal_handlerssignumZoriginal_handlerr1usr/lib/python3.10/site-packages/trio/_signals.py_signal_handler5s 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_pendingtriolowlevelZ ParkingLot_lotr _conflict_detector_closedselfrrr__init__Es   zSignalReceiver.__init__rintcCs*|jr t|dSd|j|<|jdSr)r%rr r#Zunpark)r'rrrr_addNs  zSignalReceiver._addcs d_dfdd dS)NTrrcs:jrjjdd\}}z t|WdSwdS)NFlast)r popitemr)r_ deliver_nextr'rrr0]s  z9SignalReceiver._redeliver_remaining..deliver_nextrr)r%r&rr/r_redeliver_remainingUs z#SignalReceiver._redeliver_remainingrcCs|Srrr&rrr __aiter__gszSignalReceiver.__aiter__csx|jrtd|j'|js|jIdHntjIdH|jj dd\}}|WdS1s5wYdS)Nz)open_signal_receiver block already exitedFr+) r% RuntimeErrorr$r r#Zparkr!r"Z checkpointr-)r'rr.rrr __anext__js$zSignalReceiver.__anext__Nr1)rr)rr)rr)rr))__name__ __module__ __qualname__r(r*r2r3r5rrrrrDs    rrecAsyncIterator[int]r)cCst|tsJt|jS)z/Helper for tests, not public or otherwise used.) isinstancerlenr )r9rrrget_pending_signal_countys r=signal.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 | Nonerrcsjj|dddS)NT)Z idempotent)Z run_sync_soonr*)rr@queuetokenrrrsz%open_signal_receiver..handlerN)rr)r@rArr) TypeErrorr r4r!r"Zcurrent_trio_tokenrrr2)rrrrBropen_signal_receivers"&   rF)rrrrrr)r9r:rr))rr>rr?) __future__rr collectionsrcollections.abcrrrr contextlibrtypesr typingr r!Z_utilr r rtyping_extensionsrrrr=rFrrrrs"      &  5