@OOf]dZddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl m Z dd l m Z dd l mZe d Zed Zdd ZGddZGdde eeZGdde eeZddZddZy)aq This module provides signals, which are a simple dispatching system that allows any number of interested parties to subscribe to events ("signals"). This is similar to the Blinker library (https://pypi.org/project/blinker/), with the following changes: - provides only a small subset of Blinker's functionality - supports type hints - supports async receivers. ) annotationsN) Awaitable)Callable)Any)cast)Generic) ParamSpec)TypeVarPRct|dr-ttjtj|Stj|S)zM Like weakref.ref(), but using weakref.WeakMethod for bound methods. __self__)hasattrrweakrefref WeakMethod)objs X/var/www/premiumrankchecker/venv/lib/python3.12/site-packages/mitmproxy/utils/signals.py make_weak_refrs9sJGKK!3!3C!899{{3c*eZdZddZddZddZdZy) _SignalMixincg|_yN receivers)selfs r__init__z_SignalMixin.__init__)s 68rcPt|}|jj|y)z{ Register a signal receiver. The signal will only hold a weak reference to the receiver function. N)rrappend)rreceivers rconnectz_SignalMixin.connect,s !* h'rcb|jDcgc]}||k7s|c}|_ycc}wrr)rr!rs r disconnectz_SignalMixin.disconnect5s#%)^^Gqsh!GGs,,c/Kd}|jD]}|}| ||i|d}|r(|jDcgc] }| |c}|_yycc}ww)NFTr)rargskwargscleanuprr$s rnotifyz_SignalMixin.notify8sh>> CA}(((   )-KA13?aKDN Ks:A A A AN)returnNone)r!rr+r,)__name__ __module__ __qualname__rr"r%r*rrrr(s9(H Lrrc8eZdZdfd Zdfd Zdfd ZxZS) _SyncSignalcRtj|rJt| |yr)asyncioiscoroutinefunctionsuperr"rr! __class__s rr"z_SyncSignal.connectEs#..x888 !rc$t||yrr6r%r7s rr%z_SyncSignal.disconnectI 8$rcbt||i|D]}|tj|sJyr)r6r*inspect isawaitable)rr'r(retr8s rsendz_SyncSignal.sendLs87>4262 ?C;g&9&9#&> >> ?r)r!Callable[P, None]r+r,r'zP.argsr(zP.kwargsr+r,r-r.r/r"r%r@ __classcell__r8s@rr2r2Ds"%??rr2c8eZdZdfd Zdfd Zdfd ZxZS) _AsyncSignalc$t||yr)r6r"r7s rr"z_AsyncSignal.connectRs !rc$t||yrr:r7s rr%z_AsyncSignal.disconnectUr;rcKtjt| |i|Dcgc]}|t j |r|c}d{ycc}w7 wr)r4gatherr6r*r=r>)rr'r(awsr8s rr@z_AsyncSignal.sendXs_nn!7>4:6:?w':':3'?    s!A AA A A)r!#Callable[P, Awaitable[None] | None]r+r,rBrCrEs@rrGrGQs"%  rrGc@ttttS)aK Create a synchronous signal with the given function signature for receivers. Example: s = SyncSignal(lambda event: None) # all receivers must accept a single "event" argument. def receiver(event): print(event) s.connect(receiver) s.send("foo") # prints foo s.send(event="bar") # prints bar def receiver2(): ... s.connect(receiver2) # mypy complains about receiver2 not having the right signature s2 = SyncSignal(lambda: None) # this signal has no arguments s2.send() )rr2r  receiver_specs r SyncSignalrQcs,  A ..rc@ttttS)a Create an signal that supports both regular and async receivers: Example: s = AsyncSignal(lambda event: None) async def receiver(event): print(event) s.connect(receiver) await s.send("foo") # prints foo )rrGr rOs r AsyncSignalrS}s  Q 00r)rrr+zweakref.ReferenceType)rPrAr+z_SyncSignal[P])rPrMr+z_AsyncSignal[P])__doc__ __future__rr4r=rcollections.abcrrtypingrrrr r r r rrr2rGrQrSr0rrrXs~#%$ cN CL LL8 ?'!*l ? 71:| $/4 1r