o >hՖ@sddlZddlZddlmZddlZddlmZddlmZ ddl mZ ddl m Z mZmZddlmZmZddlmZmZddlmZd Zdd d Zdd d ZGdddeZGdd d eZdddZdS)N)partial)ObservableMixin) parse_url)ComponentConfigSubscribeOptionsRegisterOptions)SessionNotReadyApplicationError)create_authenticatorIAuthenticator) SERID_TO_SER ComponentcCs(|r||n t|tstdt|trd|vrtd|ddvr,td|d|D] }|dvr=td|q0|ddkrfd D] }||vrStd |qFd D] }||vrctd |qVdS|dd krd D] }||vr{td|qndD] }||vrtd|q~dSJddS)z9 Check a WAMP connecting endpoint configuration. z'endpoint' must be a dicttypez)'type' required in endpoint configuration)tcpunixzinvalid type "{}" in endpoint)rhostportpathtlstimeoutversionz*Invalid key '{}' in endpoint configurationrrrz&'{}' required in 'tcp' endpoint config)rz''{}' not valid in 'tcp' endpoint configrz('{}' required for 'unix' endpoint config)rrrz('{}' not valid in 'unix' endpoint configFshould not arrive hereN) isinstancedict ValueErrorformatkeys)endpointcheck_native_endpointkr!w/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/autobahn/wamp/component.py_validate_endpoint1s^       r#c Cst|tkrtdt|gd}|D] }||vr$td|qd}d|vr?|ddvr:td|d|d}nd|d<d|vrO|dkrOtd |dd }|d ur{|D] }|d vrjtd |q]d D] }||vrztd |qmt}d|vr|d}t|tstdt||d}|dkr"dD] } | |vrtd| qd|vrt|d\} } } } }}d| | | d}n |d}t ||d|vrtdd|vrt|dt t fstdt dd|dDstdt }|dD]}||vrtd|ddd|Dq|dd d!g}n}|d"krd|vrh|dd#r>t|d\}} } n|dd$rSt|d\}} } } }}nt| d%krad%| d&}n d| | d'}n|d}d|vrutd(|d ur~td)d|vrt|dttfstd*|dg}nd g}nJd,i}d-D]} | |vr|| || <qt|f||dd |||||d.|S)/a Internal helper to insert defaults and create _Transport instances. :param transport: a (possibly valid) transport configuration :type transport: dict :returns: a _Transport instance :raises: ValueError on invalid configuration zs z%_create_transport..z-Invalid serializer '{}' (expected one of: {})z, cSsg|]}t|qSr!)reprr2r!r!r"r5scborjsonr0rswsr)rr)rrrz5'serializers' is only for websocket; use 'serializer'z/'headers' not supported for rawsocket transportz'serializer' must be a stringFr)r(r)r*r+r,)kindr$rr&r-r'r.)rrrrrgetformaTr parse_ws_urlr#listtupleallr join startswith parse_rs_url RuntimeErrorr1 _Transport)index transportrvalid_transport_keysr r;r-r'r.key is_securerrresourcerparamsendpoint_configvalid_serializersserialserializer_configisSecurekwr!r!r"_create_transportjs                         rTc@sReZdZdZ        dddZd d Zd d Zd dZddZddZ dS)rFz@ Thin-wrapper for WAMP transports used by a Connection. ,?皙?NcCs| durt} ||_||_||_||_| |_| |_||_|jdkr+t|dkr+t d||_ ||_ ||_ | |_ | |_| |_d|_|dS)z Nr0z5'rawsocket' transport requires exactly one serializerF)ridxrr$rr'r.r&lenrr(r)r*r+r,r-_permanent_failurereset)selfrZr;r$rr&r(r)r*r+r,r-r'r.r!r!r"__init__ s*  z_Transport.__init__cCsd|_d|_d|_|j|_dS)zP set connection failure rates and retry-delay to initial values rN)connect_attemptsconnect_sucessesconnect_failuresr* retry_delayr^r!r!r"r]5s z_Transport.resetcCs d|_dS)z Mark this transport as failed, meaning we won't try to connect to it any longer (that is: can_reconnect() will always return False afer calling this). TN)r\rdr!r!r"failed>s z_Transport.failedcCs(|jrdS|jdkr dS|j|jdkS)NFrUTrY)r\r(r`rdr!r!r" can_reconnectFs  z_Transport.can_reconnectcCsp|jdkrdS|jdkr|j|jdkrtd|j|j|_t|j|j|j|_|j|jkr5|j|_|jS)NrrUrYzmax reconnects reached) r`r(rErcr+random normalvariater,r)rdr!r!r" next_delayMs  z_Transport.next_delaycCs t|jtr |jdSt|jS)zF returns a human-readable description of the endpoint r)rrrr6rdr!r!r"describe_endpointZs   z_Transport.describe_endpoint)rUrVrWrWrXNNN) __name__ __module__ __qualname____doc__r_r]rerfrirjr!r!r!r"rFs  (  rFc@seZdZdZdZ dddZdddZ  d d d Zd d Zd!d dZ ddZ ddZ ddZ ddZ ddZddZddZddZdS)"r z A WAMP application component. A component holds configuration for (and knows how to create) transports and sessions. NFc,dus tts Jfdd}|S)a[ A decorator as a shortcut for subscribing during on-join For example:: @component.subscribe( "some.topic", options=SubscribeOptions(match='prefix'), ) def topic(*args, **kw): print("some.topic({}, {}): event received".format(args, kw)) Nc"fdd}d|S)Nc|jdS)N)topicr' check_types) subscribesessiondetails)rsfnr'rrr!r"do_subscriptionz?Component.subscribe..decorator..do_subscriptionrBon)rxryrsr'r^rrrxr" decorator z&Component.subscribe..decorator)rr)r^rrr'rsrr!r}r"rtr zComponent.subscribecro)aZ A decorator as a shortcut for registering during on-join For example:: @component.register( "com.example.add", options=RegisterOptions(invoke='roundrobin'), ) def add(*args, **kw): print("add({}, {}): event received".format(args, kw)) Ncrp)Ncrq)N) procedurer'rs)registerru)rsrxr'urir!r"do_registrationrzz>Component.register..decorator..do_registrationrBr{)rxrrsr'r^rr~r"rrz%Component.register..decorator)rr)r^rr'rsrr!rr"rrzComponent.registerrealm1c Cs|gd|durt|std||_|dur"t|s"td||_|dur+d}t|ttfr=|} d| d} | g}nt|trE|g}g|_t |D]\} } t | tkr\d| d} n| } |j t | | |j qL|pmi|_|rt||_||_||_d|_d|_d|_d|_dS) a :param main: After a transport has been connected and a session has been established and joined to a realm, this (async) procedure will be run until it finishes -- which signals that the component has run to completion. In this case, it usually doesn't make sense to use the ``on_*`` kwargs. If you do not pass a main() procedure, the session will not be closed (unless you arrange for .leave() to be called). :type main: callable taking two args ``reactor`` and ``ISession`` :param transports: Transport configurations for creating transports. Each transport can be a WAMP URL, or a dict containing the following configuration keys: - ``type`` (optional): ``websocket`` (default) or ``rawsocket`` - ``url``: the router URL - ``endpoint`` (optional, derived from URL if not provided): - ``type``: "tcp" or "unix" - ``host``, ``port``: only for TCP - ``path``: only for unix - ``timeout``: in seconds - ``tls``: ``True`` or (under Twisted) an ``twisted.internet.ssl.IOpenSSLClientComponentCreator`` instance (such as returned from ``twisted.internet.ssl.optionsForClientTLS``) or ``CertificateOptions`` instance. - ``max_retries``: Maximum number of reconnection attempts. Unlimited if set to -1. - ``initial_retry_delay``: Initial delay for reconnection attempt in seconds (Default: 1.0s). - ``max_retry_delay``: Maximum delay for reconnection attempts in seconds (Default: 60s). - ``retry_delay_growth``: The growth factor applied to the retry delay between reconnection attempts (Default 1.5). - ``retry_delay_jitter``: A 0-argument callable that introduces nose into the delay. (Default random.random) - ``serializer`` (only for raw socket): Specify an accepted serializer (e.g. 'json', 'msgpack', 'cbor', 'ubjson', 'flatbuffers') - ``serializers``: Specify list of accepted serializers - ``options``: tbd - ``proxy``: tbd :type transports: None or str or list :param realm: the realm to join :type realm: str :param authentication: configuration of authenticators :type authentication: dict :param session_factory: if None, ``ApplicationSession`` is used, otherwise a callable taking a single ``config`` argument that is used to create a new `ApplicationSession` instance. :param is_fatal: a callable taking a single argument, an ``Exception`` instance. The callable should return ``True`` if this error is "fatal", meaning we should not try connecting to the current transport again. The default behavior (on None) is to always return ``False`` )startconnectrBreadyleave disconnectconnectfailureNz%"is_fatal" must be a callable or Nonez""main" must be a callable if givenzws://127.0.0.1:8080/wsr/)rr$F)set_valid_eventscallabler _is_fatal_entryrr1r _transports enumeraterappendrT_check_native_endpoint_authenticationsession_factory_realm_extra_delay_f_done_f_session _stopping) r^main transportsconfigrealmextraauthenticationris_fatalr$rHrZ _transportr!r!r"r_sL9      zComponent.__init__cCs|jD] }|r dSqdS)NTF)rrf)r^rHr!r!r"_can_reconnects zComponent._can_reconnectcsjdurtfdd}tj||St_fdd}tj||tjdgfddfd d fd d d }t|jS)a This starts the Component, which means it will start connecting (and re-connecting) to its configured transports. A Component runs until it is "done", which means one of: - There was a "main" function defined, and it completed successfully; - Something called ``.leave()`` on our session, and we left successfully; - ``.stop()`` was called, and completed successfully; - none of our transports were able to connect successfully (failure); :returns: a Future/Deferred which will resolve (to ``None``) when we are "done" or with an error if something went wrong. Ncst|dSN)txaioresolvearg)dr!r"_cb;szComponent._start.._cbcs d_|S)z} if the _done_f future is resolved (good or bad), we want to set it to None in our class N)rrrdr!r"_resetEsz Component._start.._resetrcs\d_jrtjddSjjdt|djjdt |dt j|dS)NzInternal error {msg}msg{tb}tb) rrrrrloginfofailure_messagedebugfailure_format_tracebackreject)failrdr!r"errorUs zComponent._start..errorcsbd_fddfddfdd}fdd}tjd }t|||dS) Ncsjjdt|djjdt|dt|jtr(jjd|j dn7t|jt r:jj dt|dn% |jrT|jj dd\}}}jjd |d n jjd t|djdurgd }n|j}|r{j d dtdddS)Nzcomponent failed: {error})rrrz{msg}rz&Connection failed with OS error: {msg}rzTLS failure: {reason})reasonzConnection failed: {error}Fz"Error was fatal; failing transport)rrrrrrvaluer r error_messageOSErrorr _is_ssl_errorargsrre call_later)rssl_libssl_func ssl_reasonr)r^transport_candidatetransport_checkr!r"handle_connect_errores*        zGComponent._start..attempt_connect..handle_connect_errorcs>tdj}t|fddfddS)Nrc tSrrr_chain_frr!r" zYComponent._start..attempt_connect..notify_connect_error..crrrrrr!r"rr)r create_futurefirer add_callbacks)r handler_frdrr"notify_connect_errors  zGComponent._start..attempt_connect..notify_connect_errorcs|}t|ddSr)rr)rnotify_frrr!r" connect_errorsz@Component._start..attempt_connect..connect_errorcstjddSr)rrr)xrdr!r" session_donerzz?Component._start..attempt_connect..session_doner)rr as_future _connect_oncer)rrr connect_f)loopr^rrrr"attempt_connectbs 0 z)Component._start..attempt_connectc sjds1d}j|zt|ty0}ztj|WYd}~dSd}~ww t}| r?|d<nq2| }jj d|j |j |dt|_tjdS)NzEntering re-connect loopz:Component failed: Exhausted all transport connect attemptsTrzZtrying transport {transport_idx} ("{transport_url}") using connect delay {transport_delay}) transport_idx transport_urltransport_delay)rrrrrErrrnextrfriwarnrZr$sleeprr)rerr_msgerHdelay)rrr^r transport_genr!r"rs4   z)Component._start..transport_checkr)rrrr itertoolscyclerr)r^rrrstart_fr!)rrrrr^rrrr"_start&s       N zComponent._startcCsZd|_|jr|jr|jS|jrttj|jSt|j s(t |j dt dS)NT) rr is_attachedrrrrcancel is_calledrrcreate_future_successrdr!r!r"stops   zComponent.stopcstjjdjdtfdd}jd7_tj|}fdd}t |d|S)NzWconnecting once using transport type "{transport_type}" over endpoint "{endpoint_desc}")transport_type endpoint_descc stjj}z,|_}jD]\}}t|tr$| |qt |fi|}| |qWnt yK}z t |}t |d}~ww|_fdd}|d|fdd}jdurp|d|fdd} |d | |S) Ncs\jjd|dts,|jdvrtddStt|j|j}t |dSdS)Nz"session leaving '{details.reason}'rw)zwamp.close.normalzwamp.close.goodbye_and_out) rrrrrrcreate_failurer messager)rvrwfdoner^r!r"on_leave s   zAComponent._connect_once..create_session..on_leavercsfjd7_jjd|dtj}fdd}fdd}t|||dS)NrYzsession on_join: {details}rcs(jdfdd}td|dS)N main_successcs$zWdStyYdSwr)rrr!rvr!r"r's  zeComponent._connect_once..create_session..on_join..main_success..leaver)rrrr)rr)r^rvr!r"r$s  zVComponent._connect_once..create_session..on_join..main_successcs(jjd|dt|dS)Nzmain_error: {err}err)rrrrrr)rr^rvr!r" main_error0s  zTComponent._connect_once..create_session..on_join..main_error)r]rarrrrrr)rvrwrrrrreactorr^rHrr"on_joins z@Component._connect_once..create_session..on_joinrBcsBjjd|dts|sjddStddSdS)Nz,session on_disconnect: was_clean={was_clean}) was_cleanzSession disconnected uncleanly)rrrrrr)rvrrr!r" on_disconnect;s zFComponent._connect_once..create_session..on_disconnectr)rrrrrritemsrr add_authenticatorr Exceptionrrr_parentr|r) cfgrv auth_name auth_config authenticatorrrrrrrr!r"create_sessions0          z/Component._connect_once..create_sessionrYcs,jd7_tst|dSdS)a this may seem redundant after looking at _connect_transport, but it will handle a case where something goes wrong in _connect_transport itself -- as the only connect our caller has is the 'done' future rYN)rbrrrr)rrHr!r"on_errorUs z)Component._connect_once..on_error) rrrrjrrr`r_connect_transportr)r^rrHrrrr!rr"rs^ zComponent._connect_oncecC|d|dS)z A decorator as a shortcut for listening for 'join' events. For example:: @component.on_join def joined(session, details): print("Session {} joined: {}".format(session, details)) rBNr{r^rxr!r!r"res zComponent.on_joincCr)zM A decorator as a shortcut for listening for 'leave' events. rNr{rr!r!r"rqzComponent.on_leavecCr)zO A decorator as a shortcut for listening for 'connect' events. rNr{rr!r!r" on_connectwrzComponent.on_connectcCr)zR A decorator as a shortcut for listening for 'disconnect' events. rNr{rr!r!r"r}rzComponent.on_disconnectcCr)zM A decorator as a shortcut for listening for 'ready' events. rNr{rr!r!r"on_readyrzComponent.on_readycCr)zV A decorator as a shortcut for listening for 'connectfailure' events. rNr{rr!r!r"on_connectfailurerzComponent.on_connectfailure)NF)NNNrNNNNr)rkrlrmrnrrtrr_rrrrrrrrrrr!r!r!r"r gs*    /  c st|tr|g}t|tkrtdt||D]}t|ts)tdt|qtfddfddfdd}g}|D] }||}||qFtj |d d }rjfd d } t || | |S) a Internal helper. Use "run" method from autobahn.twisted.wamp or autobahn.asyncio.wamp This is the generic parts of the run() method so that there's very little code in the twisted/asyncio specific run() methods. This is called by react() (or run_until_complete() so any errors coming out of this should be handled properly. Logging will already be started. zB"components" must be a list of Component objects - encountered {0}zN"components" must be a list of Component objects - encountereditem of type {0}csjd||d|S)Nz-Component '{c}' successfully completed: {arg})crr)comprrr!r"component_successsz_run..component_successcs.jd|t|djdt|ddS)NzComponent '{c}' error: {msg})rrzComponent error: {tb}r)rrrrr)rrrr!r"component_failuresz_run..component_failurecs,t|j}t|t|t||Sr)rrrrr)rr)rrrr!r"component_startsz_run..component_startF)consume_exceptionscsd|dS)Nz&All components ended; stopping reactorrr) done_callbackrrr!r"all_dones z_run..all_done) rr rr?rrr make_loggerrgatherr) r componentsrrrdlrrdone_dr r!)rrrrrr"_runs6        r&r)rrg functoolsrr autobahn.utilrautobahn.websocket.utilrr>autobahn.rawsocket.utilrDautobahn.wamp.typesrrrautobahn.wamp.exceptionrr autobahn.wamp.authr r autobahn.wamp.serializerr __all__r#rTobjectrFr r&r!r!r!r"s.       9_-