U g@sddlmZddlZddlmZddlmZddlmZddl Z ddl m Z m Z m Z erddlmZmZmZmZdd lmZdd lmZed d d dddZGdddZdddddZedddddZdS)) annotationsN) OrderedDict)contextmanager) TYPE_CHECKING)ConflictDetectoris_main_thread signal_raise) AsyncIteratorCallable GeneratorIterable) FrameType)Selfz Iterable[int]zHCallable[[int, FrameType | None], object] | int | signal.Handlers | NonezGenerator[(None, None, None)])signalshandlerreturnc csRi}z(t|D]}t||||<qdVW5|D]\}}t||q6XdSN)itemssignalset)rrZoriginal_handlerssignumoriginal_handlerr1/tmp/pip-unpacked-wheel-ks04xdmi/trio/_signals.py_signal_handler6s  rc@sTeZdZddddZdddddZddd d Zd dd d ZddddZdS)SignalReceiverNonercCs(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__Fs  zSignalReceiver.__init__int)rrcCs(|jrt|nd|j|<|jdSr)r$r rr"Zunpark)r&rrrr_addOs  zSignalReceiver._addcs$d_ddfdd dS)NTrrcs2jr.jjdd\}}z t|W5XdS)NFlast)rpopitemr )r_ deliver_nextr&rrr/^s  z9SignalReceiver._redeliver_remaining..deliver_next)r$r%rr.r_redeliver_remainingVsz#SignalReceiver._redeliver_remainingrcCs|Srrr%rrr __aiter__hszSignalReceiver.__aiter__c sj|jrtd|jL|js.|jIdHntjIdH|jj dd\}}|W5QRSQRXdS)Nz)open_signal_receiver block already exitedFr*) r$ RuntimeErrorr#rr"Zparkr r! checkpointr,)r&rr-rrr __anext__kszSignalReceiver.__anext__N)__name__ __module__ __qualname__r'r)r0r1r4rrrrrEs  rzAsyncIterator[int]r()recrcCst|tstt|jS)z/Helper for tests, not public or otherwise used.) isinstancerAssertionErrorlenr)r8rrrget_pending_signal_countzsr<zsignal.Signals | intz)Generator[AsyncIterator[int], None, None])rrc 'sr|s tdtstdtjtddddfdd }z t|| VW5QRXW5Xd S) 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 threadr(zFrameType | Noner)rframercsjj|dddS)NT)Z idempotent)Z run_sync_soonr))rr=queuetokenrrrsz%open_signal_receiver..handlerN) TypeErrorrr2r r!Zcurrent_trio_tokenrr0r)rrrr>ropen_signal_receivers&  rB) __future__rr collectionsr contextlibrtypingrr Z_utilrrr collections.abcr r r r typesrZtyping_extensionsrrrr<rBrrrrs      &5