U ucܕ@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||nt|ts tdt|tr*d|krsz%_create_transport..z-Invalid serializer '{}' (expected one of: {})z, cSsg|] }t|qSr")reprr1r"r"r#r4sZcborjsonr/Zrswsr)rr)rrrz5'serializers' is only for websocket; use 'serializer'z'serializer' must be a stringFr)r(r)r*r+r,)kindr%rr&r-r')rrrrrgetZformaTr parse_ws_urlr$listtupleallr join startswith parse_rs_url RuntimeErrorr0r _Transport)index transportr Zvalid_transport_keysr!r8r-r'keyZ is_securerrresourcerparamsZendpoint_configZvalid_serializersserialZserializer_configZisSecurekwr"r"r#_create_transportjs                            rJc@sBeZdZdZdddZd d Zd d Zd dZddZddZ dS)rBz@ Thin-wrapper for WAMP transports used by a Connection. ,?皙?Nc Cs| dkrt} ||_||_||_||_| |_||_|jdkrPt|dkrPtd||_ ||_ ||_ | |_ | |_ | |_d|_|dS)z Nr/z5'rawsocket' transport requires exactly one serializerF)ridxrr%rr'r&lenrr(r)r*r+r,r-_permanent_failurereset) selfrPr8r%rr&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_delayrTr"r"r#rS.sz_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)rRrZr"r"r#failed7sz_Transport.failedcCs(|jr dS|jdkrdS|j|jdkS)NFrKTrO)rRr(rVrZr"r"r# can_reconnect?s  z_Transport.can_reconnectcCsv|jdkrdS|jdkr2|j|jdkr2tdn@|j|j|_t|j|j|j|_|j|jkrl|j|_|jSdS)NrrKrOzmax reconnects reached) rVr(rArYr+random normalvariater,r)rZr"r"r# next_delayFs   z_Transport.next_delaycCs t|jtr|jdSt|jS)zF returns a human-readable description of the endpoint r)rrrr5rZr"r"r#describe_endpointSs  z_Transport.describe_endpoint)rKrLrMrMrNNN) __name__ __module__ __qualname____doc__rUrSr[r\r_r`r"r"r"r#rBs &  rBc@seZdZdZdZdddZd ddZd!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. NFcs,dksttstfdd}|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)) Ncs"fdd}d|S)Ncs|jdS)N)topicr' check_types) subscribesessiondetails)rffnr'rer"r#do_subscription|sz?Component.subscribe..decorator..do_subscriptionr>on)rkrlrfr'rTrerkr# decoratorzs z&Component.subscribe..decorator)rrr)rTrer'rfrqr"ror#rgks zComponent.subscribecs,dksttstfdd}|S)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)) Ncs"fdd}d|S)Ncs|jdS)N)Z procedurer'rf)registerrh)rfrkr'urir"r#do_registrationsz>Component.register..decorator..do_registrationr>rm)rkrtrfr'rTrsrpr#rqs z%Component.register..decorator)rrr)rTrsr'rfrqr"rur#rrs zComponent.registerrealm1c Cs|dddddddg|dk r0t|s0td ||_|dk rNt|sNtd ||_|dkr`d }t|ttfr|} d | d } | g}nt|tr|g}g|_t |D]:\} } t | tkrd | d } n| } |j t | | |j q|pi|_|r||_||_||_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`` startconnectr>readyleave disconnectconnectfailureNz%"is_fatal" must be a callable or Nonez""main" must be a callable if givenzws://127.0.0.1:8080/wsr.)rr%F)Zset_valid_eventscallabler _is_fatal_entryrr0r _transports enumeraterappendrJZ_check_native_endpoint_authenticationsession_factory_realm_extra_delay_f_done_f_session _stopping) rTmain transportsconfigrealmextraZauthenticationris_fatalr%rDrP _transportr"r"r#rUsZ9     zComponent.__init__cCs|jD]}|rdSqdS)NTF)rr\)rTrDr"r"r#_can_reconnects zComponent._can_reconnectcsjdk r2tfdd}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#_cb4szComponent._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)rrrZr"r#_reset>sz Component._start.._resetrcsZd_jrtjdn:jjdt|djjdt |dt j|dS)NzInternal error {msg}msg{tb}tb) rrrrrloginfofailure_messagedebugfailure_format_tracebackreject)failrZr"r#errorNs zComponent._start..errorcsbd_fddfddfdd}fdd}tjd }t|||dS) Ncsjjdt|djjdt|dt|jtrPjjd|j dnnt|jt rtjj dt|dnJ |jr|jj dd\}}}jjd |d njjd t|djdkrd }n |j}|rj 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_messageOSErrorrZ _is_ssl_errorargsr~r[ call_later)rZssl_libZssl_funcZ ssl_reasonr)rTtransport_candidatetransport_checkr"r#handle_connect_error^s*       zGComponent._start..attempt_connect..handle_connect_errorcs>tdj}t|fddfddS)Nr|cs tSrrr_Zchain_frr"r#zYComponent._start..attempt_connect..notify_connect_error..cs tSrrrrr"r#rr)r create_futurefirer add_callbacks)rZ handler_frZrr#notify_connect_errors  zGComponent._start..attempt_connect..notify_connect_errorcs|}t|ddSr)rr)rZnotify_frrr"r# connect_errorsz@Component._start..attempt_connect..connect_errorcstjddSr)rrr)xrZr"r# session_donesz?Component._start..attempt_connect..session_doner)rr as_future _connect_oncer)rrrZ connect_f)looprTrrrr#attempt_connect[s0  z)Component._start..attempt_connectc sjdshd}j|z t|Wn6tk rf}ztj|WYdSd}~XYnXt}| rh|d<qqh| }jj d|j |j |dt|_tjdS)NzEntering re-connect loopz:Component failed: Exhausted all transport connect attemptsrzZtrying transport {transport_idx} ("{transport_url}") using connect delay {transport_delay})Z transport_idxZ transport_urlZtransport_delay)rrrrrArrrnextr\r_warnrPr%sleeprr)rerr_msgerDdelay)rrrTr transport_genr"r#rs,    z)Component._start..transport_checkrw)rrrr itertoolscyclerr)rTrrrZstart_fr")rrrrrTrrrr#_starts       N zComponent._startcCsZd|_|jr |jr |jS|jr6ttj|jSt|j sPt |j dt dS)NT) rrZ is_attachedrzrrrcancel is_calledrrZcreate_future_successrZr"r"r#stops  zComponent.stopcstjjdjdtfdd}jd7_tj|}fdd}t |d|S)NzWconnecting once using transport type "{transport_type}" over endpoint "{endpoint_desc}")Ztransport_typeZ endpoint_descc stjj}zT|_}jD]4\}}t|trH| |q*t |f|}| |q*Wn:t k r}zt |}t |W5d}~XYnhX|_fdd}|d|fdd}jdk r|d|fdd} |d | |SdS) NcsVjjd|dtsR|jdkr2tdn tt|j|j}t |dS)Nz"session leaving '{details.reason}'rj)zwamp.close.normalzwamp.close.goodbye_and_out) rrrrrrcreate_failurer messager)rirjfdonerTr"r#on_leaves   zAComponent._connect_once..create_session..on_leaverzcsfjd7_jjd|dtj}fdd}fdd}t|||dS)NrOzsession on_join: {details}rcs(jdfdd}td|dS)N main_successcs&z Wntk r YnXdSr)rzrr"rir"r#rz s zeComponent._connect_once..create_session..on_join..main_success..leaver)rrrr)rrz)rTrir"r#rs  zVComponent._connect_once..create_session..on_join..main_successcs(jjd|dt|dS)Nzmain_error: {err}err)rrrrr{r)rrTrir"r# main_error)s zTComponent._connect_once..create_session..on_join..main_error)rSrWrrrrrr)rirjrrrrreactorrTrDrr#on_joins z@Component._connect_once..create_session..on_joinr>cs<jjd|dts8|s,jdn tddS)Nz,session on_disconnect: was_clean={was_clean}) was_cleanzSession disconnected uncleanly)rrrrrr)rirrr"r# on_disconnect4s zFComponent._connect_once..create_session..on_disconnectr{)rrrrrritemsrr Zadd_authenticatorr ExceptionrrrZ_parentrnr) ZcfgriZ auth_nameZ auth_configZ authenticatorrrrrrrr"r#create_sessions*         z/Component._connect_once..create_sessionrOcs(jd7_ts$t|dS)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 rON)rXrrrr)rrDr"r#on_errorNs z)Component._connect_once..on_error) rrrr`rrrVrZ_connect_transportr)rTrrDrrrr"rr#rs$^ zComponent._connect_oncecCs|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)) r>NrmrTrkr"r"r#r^s zComponent.on_joincCs|d|dS)zM A decorator as a shortcut for listening for 'leave' events. rzNrmrr"r"r#rjszComponent.on_leavecCs|d|dS)zO A decorator as a shortcut for listening for 'connect' events. rxNrmrr"r"r# on_connectpszComponent.on_connectcCs|d|dS)zR A decorator as a shortcut for listening for 'disconnect' events. r{Nrmrr"r"r#rvszComponent.on_disconnectcCs|d|dS)zM A decorator as a shortcut for listening for 'ready' events. ryNrmrr"r"r#on_ready|szComponent.on_readycCs|d|dS)zV A decorator as a shortcut for listening for 'connectfailure' events. r|Nrmrr"r"r#on_connectfailureszComponent.on_connectfailure)NF)NF)NNNrvNNNN)N)rarbrcrdrrgrrrUrrrrrrrrrrr"r"r"r#r `s*    / c st|tr|g}t|tkr.tdt||D] }t|ts2tdt|q2tfddfddfdd}g}|D]}||}||qtj |d d }rԇfd 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)rrrwrr)rr)rrrr"r#component_startsz_run..component_startF)Zconsume_exceptionscsd|dS)Nz&All components ended; stopping reactorrr) done_callbackrrr"r#all_dones z_run..all_done) rr rr;rrrZ make_loggerrgatherr) r componentsrrrZdlrrZdone_drr")rrrrrr#_runs8      r)N)N)N)rr] functoolsrrZ autobahn.utilrZautobahn.websocket.utilrr:Zautobahn.rawsocket.utilr@Zautobahn.wamp.typesrrrZautobahn.wamp.exceptionrr Zautobahn.wamp.authr r Zautobahn.wamp.serializerr __all__r$rJobjectrBr rr"r"r"r#s,      9 ]-