Source code for autobahn.wamp.protocol

###############################################################################
##
##  Copyright (C) 2013-2014 Tavendo GmbH
##
##  Licensed under the Apache License, Version 2.0 (the "License");
##  you may not use this file except in compliance with the License.
##  You may obtain a copy of the License at
##
##      http://www.apache.org/licenses/LICENSE-2.0
##
##  Unless required by applicable law or agreed to in writing, software
##  distributed under the License is distributed on an "AS IS" BASIS,
##  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
##  See the License for the specific language governing permissions and
##  limitations under the License.
##
###############################################################################

from __future__ import absolute_import

import inspect
import six
from six import StringIO

from autobahn.wamp.interfaces import ISession, \
                                     IPublication, \
                                     IPublisher, \
                                     ISubscription, \
                                     ISubscriber, \
                                     ICaller, \
                                     IRegistration, \
                                     ITransportHandler

from autobahn import util
from autobahn import wamp
from autobahn.wamp import uri
from autobahn.wamp import message
from autobahn.wamp import types
from autobahn.wamp import role
from autobahn.wamp import exception
from autobahn.wamp.exception import ProtocolError, SessionNotReady
from autobahn.wamp.types import SessionDetails



[docs]class Endpoint: """ """ def __init__(self, obj, fn, procedure, options = None): self.obj = obj self.fn = fn self.procedure = procedure self.options = options
[docs]class Handler: """ """ def __init__(self, obj, fn, topic, details_arg = None): self.obj = obj self.fn = fn self.topic = topic self.details_arg = details_arg
[docs]class Publication: """ Object representing a publication. This class implements :class:`autobahn.wamp.interfaces.IPublication`. """ def __init__(self, publicationId): self.id = publicationId
IPublication.register(Publication)
[docs]class Subscription: """ Object representing a subscription. This class implements :class:`autobahn.wamp.interfaces.ISubscription`. """ def __init__(self, session, subscriptionId): self._session = session self.active = True self.id = subscriptionId
[docs] def unsubscribe(self): """ Implements :func:`autobahn.wamp.interfaces.ISubscription.unsubscribe` """ return self._session._unsubscribe(self)
ISubscription.register(Subscription)
[docs]class Registration: """ Object representing a registration. This class implements :class:`autobahn.wamp.interfaces.IRegistration`. """ def __init__(self, session, registrationId): self._session = session self.active = True self.id = registrationId
[docs] def unregister(self): """ Implements :func:`autobahn.wamp.interfaces.IRegistration.unregister` """ return self._session._unregister(self)
IRegistration.register(Registration)
[docs]class BaseSession: """ WAMP session base class. This class implements :class:`autobahn.wamp.interfaces.ISession`. """ def __init__(self): """ """ ## this is for library level debugging self.debug = False ## this is for app level debugging. exceptions raised in user code ## will get logged (that is, when invoking remoted procedures or ## when invoking event handlers) self.debug_app = False ## this is for marshalling traceback from exceptions thrown in user ## code within WAMP error messages (that is, when invoking remoted ## procedures) self.traceback_app = False ## mapping of exception classes to WAMP error URIs self._ecls_to_uri_pat = {} ## mapping of WAMP error URIs to exception classes self._uri_to_ecls = {} ## session authentication information ## self._authid = None self._authrole = None self._authmethod = None self._authprovider = None
[docs] def onConnect(self): """ Implements :func:`autobahn.wamp.interfaces.ISession.onConnect` """
[docs] def onJoin(self, details): """ Implements :func:`autobahn.wamp.interfaces.ISession.onJoin` """
[docs] def onLeave(self, details): """ Implements :func:`autobahn.wamp.interfaces.ISession.onLeave` """
[docs] def onDisconnect(self): """ Implements :func:`autobahn.wamp.interfaces.ISession.onDisconnect` """
[docs] def define(self, exception, error = None): """ Implements :func:`autobahn.wamp.interfaces.ISession.define` """ if error is None: assert(hasattr(exception, '_wampuris')) self._ecls_to_uri_pat[exception] = exception._wampuris self._uri_to_ecls[exception._wampuris[0].uri()] = exception else: assert(not hasattr(exception, '_wampuris')) self._ecls_to_uri_pat[exception] = [uri.Pattern(six.u(error), uri.Pattern.URI_TARGET_HANDLER)] self._uri_to_ecls[six.u(error)] = exception
def _message_from_exception(self, request_type, request, exc, tb = None): """ Create a WAMP error message from an exception. :param request_type: The request type this WAMP error message is for. :type request_type: int :param request: The request ID this WAMP error message is for. :type request: int :param exc: The exception. :type exc: Instance of :class:`Exception` or subclass thereof. :param tb: Optional traceback. If present, it'll be included with the WAMP error message. :type tb: list or None """ args = None if hasattr(exc, 'args'): args = list(exc.args) # make sure tuples are made into lists kwargs = None if hasattr(exc, 'kwargs'): kwargs = exc.kwargs if tb: if kwargs: kwargs['traceback'] = tb else: kwargs = {'traceback': tb} if isinstance(exc, exception.ApplicationError): error = exc.error if type(exc.error) == six.text_type else six.u(exc.error) else: if exc.__class__ in self._ecls_to_uri_pat: error = self._ecls_to_uri_pat[exc.__class__][0]._uri else: error = u"wamp.error.runtime_error" msg = message.Error(request_type, request, error, args, kwargs) return msg def _exception_from_message(self, msg): """ Create a user (or generic) exception from a WAMP error message. :param msg: A WAMP error message. :type msg: instance of :class:`autobahn.wamp.message.Error` """ # FIXME: # 1. map to ecls based on error URI wildcard/prefix # 2. extract additional args/kwargs from error URI exc = None if msg.error in self._uri_to_ecls: ecls = self._uri_to_ecls[msg.error] try: ## the following might fail, eg. TypeError when ## signature of exception constructor is incompatible ## with args/kwargs or when the exception constructor raises if msg.kwargs: if msg.args: exc = ecls(*msg.args, **msg.kwargs) else: exc = ecls(**msg.kwargs) else: if msg.args: exc = ecls(*msg.args) else: exc = ecls() except Exception: ## FIXME: log e pass if not exc: ## the following ctor never fails .. if msg.kwargs: if msg.args: exc = exception.ApplicationError(msg.error, *msg.args, **msg.kwargs) else: exc = exception.ApplicationError(msg.error, **msg.kwargs) else: if msg.args: exc = exception.ApplicationError(msg.error, *msg.args) else: exc = exception.ApplicationError(msg.error) return exc
ISession.register(BaseSession)
[docs]class ApplicationSession(BaseSession): """ WAMP endpoint session. This class implements * :class:`autobahn.wamp.interfaces.IPublisher` * :class:`autobahn.wamp.interfaces.ISubscriber` * :class:`autobahn.wamp.interfaces.ICaller` * :class:`autobahn.wamp.interfaces.ICallee` * :class:`autobahn.wamp.interfaces.ITransportHandler` """ def __init__(self, config = None): """ Constructor. """ BaseSession.__init__(self) self.config = config or types.ComponentConfig(realm = u"default") self._transport = None self._session_id = None self._realm = None self._session_id = None self._goodbye_sent = False self._transport_is_closing = False ## outstanding requests self._publish_reqs = {} self._subscribe_reqs = {} self._unsubscribe_reqs = {} self._call_reqs = {} self._register_reqs = {} self._unregister_reqs = {} ## subscriptions in place self._subscriptions = {} ## registrations in place self._registrations = {} ## incoming invocations self._invocations = {}
[docs] def onOpen(self, transport): """ Implements :func:`autobahn.wamp.interfaces.ITransportHandler.onOpen` """ self._transport = transport self.onConnect()
[docs] def onConnect(self): """ Implements :func:`autobahn.wamp.interfaces.ISession.onConnect` """ self.join(self.config.realm)
[docs] def join(self, realm, authmethods = None, authid = None): """ Implements :func:`autobahn.wamp.interfaces.ISession.join` """ if six.PY2 and type(realm) == str: realm = six.u(realm) if self._session_id: raise Exception("already joined") self._goodbye_sent = False roles = [ role.RolePublisherFeatures(), role.RoleSubscriberFeatures(), role.RoleCallerFeatures(), role.RoleCalleeFeatures() ] msg = message.Hello(realm, roles, authmethods, authid) self._realm = realm self._transport.send(msg)
[docs] def disconnect(self): """ Implements :func:`autobahn.wamp.interfaces.ISession.disconnect` """ if self._transport: self._transport.close() else: raise Exception("transport disconnected")
[docs] def onMessage(self, msg): """ Implements :func:`autobahn.wamp.interfaces.ITransportHandler.onMessage` """ if self._session_id is None: ## the first message must be WELCOME, ABORT or CHALLENGE .. ## if isinstance(msg, message.Welcome): self._session_id = msg.session details = SessionDetails(self._realm, self._session_id, msg.authid, msg.authrole, msg.authmethod) self._as_future(self.onJoin, details) elif isinstance(msg, message.Abort): ## fire callback and close the transport self.onLeave(types.CloseDetails(msg.reason, msg.message)) elif isinstance(msg, message.Challenge): challenge = types.Challenge(msg.method, msg.extra) d = self._as_future(self.onChallenge, challenge) def success(signature): reply = message.Authenticate(signature) self._transport.send(reply) def error(err): reply = message.Abort(u"wamp.error.cannot_authenticate", u"{0}".format(err.value)) self._transport.send(reply) ## fire callback and close the transport self.onLeave(types.CloseDetails(reply.reason, reply.message)) self._add_future_callbacks(d, success, error) else: raise ProtocolError("Received {0} message, and session is not yet established".format(msg.__class__)) else: if isinstance(msg, message.Goodbye): if not self._goodbye_sent: ## the peer wants to close: send GOODBYE reply reply = message.Goodbye() self._transport.send(reply) self._session_id = None ## fire callback and close the transport self.onLeave(types.CloseDetails(msg.reason, msg.message)) ## consumer messages ## elif isinstance(msg, message.Event): if msg.subscription in self._subscriptions: handler = self._subscriptions[msg.subscription] if handler.details_arg: if not msg.kwargs: msg.kwargs = {} msg.kwargs[handler.details_arg] = types.EventDetails(publication = msg.publication, publisher = msg.publisher) try: if handler.obj: if msg.kwargs: if msg.args: handler.fn(handler.obj, *msg.args, **msg.kwargs) else: handler.fn(handler.obj, **msg.kwargs) else: if msg.args: handler.fn(handler.obj, *msg.args) else: handler.fn(handler.obj) else: if msg.kwargs: if msg.args: handler.fn(*msg.args, **msg.kwargs) else: handler.fn(**msg.kwargs) else: if msg.args: handler.fn(*msg.args) else: handler.fn() except Exception as e: if self.debug_app: print("Failure while firing event handler {0} subscribed under '{1}' ({2}): {3}".format(handler.fn, handler.topic, msg.subscription, e)) else: raise ProtocolError("EVENT received for non-subscribed subscription ID {0}".format(msg.subscription)) elif isinstance(msg, message.Published): if msg.request in self._publish_reqs: d, opts = self._publish_reqs.pop(msg.request) p = Publication(msg.publication) self._resolve_future(d, p) else: raise ProtocolError("PUBLISHED received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Subscribed): if msg.request in self._subscribe_reqs: d, obj, fn, topic, options = self._subscribe_reqs.pop(msg.request) if options: self._subscriptions[msg.subscription] = Handler(obj, fn, topic, options.details_arg) else: self._subscriptions[msg.subscription] = Handler(obj, fn, topic) s = Subscription(self, msg.subscription) self._resolve_future(d, s) else: raise ProtocolError("SUBSCRIBED received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Unsubscribed): if msg.request in self._unsubscribe_reqs: d, subscription = self._unsubscribe_reqs.pop(msg.request) if subscription.id in self._subscriptions: del self._subscriptions[subscription.id] subscription.active = False self._resolve_future(d, None) else: raise ProtocolError("UNSUBSCRIBED received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Result): if msg.request in self._call_reqs: if msg.progress: ## progressive result ## _, opts = self._call_reqs[msg.request] if opts.onProgress: try: if msg.kwargs: if msg.args: opts.onProgress(*msg.args, **msg.kwargs) else: opts.onProgress(**msg.kwargs) else: if msg.args: opts.onProgress(*msg.args) else: opts.onProgress() except Exception as e: ## silently drop exceptions raised in progressive results handlers if self.debug: print("Exception raised in progressive results handler: {0}".format(e)) else: ## silently ignore progressive results pass else: ## final result ## d, opts = self._call_reqs.pop(msg.request) if msg.kwargs: if msg.args: res = types.CallResult(*msg.args, **msg.kwargs) else: res = types.CallResult(**msg.kwargs) self._resolve_future(d, res) else: if msg.args: if len(msg.args) > 1: res = types.CallResult(*msg.args) self._resolve_future(d, res) else: self._resolve_future(d, msg.args[0]) else: self._resolve_future(d, None) else: raise ProtocolError("RESULT received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Invocation): if msg.request in self._invocations: raise ProtocolError("INVOCATION received for request ID {0} already invoked".format(msg.request)) else: if msg.registration not in self._registrations: raise ProtocolError("INVOCATION received for non-registered registration ID {0}".format(msg.registration)) else: endpoint = self._registrations[msg.registration] if endpoint.options and endpoint.options.details_arg: if not msg.kwargs: msg.kwargs = {} if msg.receive_progress: def progress(*args, **kwargs): progress_msg = message.Yield(msg.request, args = args, kwargs = kwargs, progress = True) self._transport.send(progress_msg) else: progress = None msg.kwargs[endpoint.options.details_arg] = types.CallDetails(progress, caller = msg.caller, caller_transport = msg.caller_transport, authid = msg.authid, authrole = msg.authrole, authmethod = msg.authmethod) if endpoint.obj: if msg.kwargs: if msg.args: d = self._as_future(endpoint.fn, endpoint.obj, *msg.args, **msg.kwargs) else: d = self._as_future(endpoint.fn, endpoint.obj, **msg.kwargs) else: if msg.args: d = self._as_future(endpoint.fn, endpoint.obj, *msg.args) else: d = self._as_future(endpoint.fn, endpoint.obj) else: if msg.kwargs: if msg.args: d = self._as_future(endpoint.fn, *msg.args, **msg.kwargs) else: d = self._as_future(endpoint.fn, **msg.kwargs) else: if msg.args: d = self._as_future(endpoint.fn, *msg.args) else: d = self._as_future(endpoint.fn) def success(res): del self._invocations[msg.request] if isinstance(res, types.CallResult): reply = message.Yield(msg.request, args = res.results, kwargs = res.kwresults) else: reply = message.Yield(msg.request, args = [res]) self._transport.send(reply) def error(err): if self.traceback_app: ## if asked to marshal the traceback within the WAMP error message, extract it # noinspection PyCallingNonCallable tb = StringIO() err.printTraceback(file = tb) tb = tb.getvalue().splitlines() else: tb = None if self.debug_app: print("Failure while invoking procedure {0} registered under '{1}' ({2}):".format(endpoint.fn, endpoint.procedure, msg.registration)) print(err) del self._invocations[msg.request] if hasattr(err, 'value'): exc = err.value else: exc = err reply = self._message_from_exception(message.Invocation.MESSAGE_TYPE, msg.request, exc, tb) self._transport.send(reply) self._invocations[msg.request] = d self._add_future_callbacks(d, success, error) elif isinstance(msg, message.Interrupt): if msg.request not in self._invocations: raise ProtocolError("INTERRUPT received for non-pending invocation {0}".format(msg.request)) else: # noinspection PyBroadException try: self._invocations[msg.request].cancel() except Exception: if self.debug: print("could not cancel call {0}".format(msg.request)) finally: del self._invocations[msg.request] elif isinstance(msg, message.Registered): if msg.request in self._register_reqs: d, obj, fn, procedure, options = self._register_reqs.pop(msg.request) self._registrations[msg.registration] = Endpoint(obj, fn, procedure, options) r = Registration(self, msg.registration) self._resolve_future(d, r) else: raise ProtocolError("REGISTERED received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Unregistered): if msg.request in self._unregister_reqs: d, registration = self._unregister_reqs.pop(msg.request) if registration.id in self._registrations: del self._registrations[registration.id] registration.active = False self._resolve_future(d, None) else: raise ProtocolError("UNREGISTERED received for non-pending request ID {0}".format(msg.request)) elif isinstance(msg, message.Error): d = None ## ERROR reply to PUBLISH ## if msg.request_type == message.Publish.MESSAGE_TYPE and msg.request in self._publish_reqs: d = self._publish_reqs.pop(msg.request)[0] ## ERROR reply to SUBSCRIBE ## elif msg.request_type == message.Subscribe.MESSAGE_TYPE and msg.request in self._subscribe_reqs: d = self._subscribe_reqs.pop(msg.request)[0] ## ERROR reply to UNSUBSCRIBE ## elif msg.request_type == message.Unsubscribe.MESSAGE_TYPE and msg.request in self._unsubscribe_reqs: d = self._unsubscribe_reqs.pop(msg.request)[0] ## ERROR reply to REGISTER ## elif msg.request_type == message.Register.MESSAGE_TYPE and msg.request in self._register_reqs: d = self._register_reqs.pop(msg.request)[0] ## ERROR reply to UNREGISTER ## elif msg.request_type == message.Unregister.MESSAGE_TYPE and msg.request in self._unregister_reqs: d = self._unregister_reqs.pop(msg.request)[0] ## ERROR reply to CALL ## elif msg.request_type == message.Call.MESSAGE_TYPE and msg.request in self._call_reqs: d = self._call_reqs.pop(msg.request)[0] if d: self._reject_future(d, self._exception_from_message(msg)) else: raise ProtocolError("WampAppSession.onMessage(): ERROR received for non-pending request_type {0} and request ID {1}".format(msg.request_type, msg.request)) elif isinstance(msg, message.Heartbeat): pass ## FIXME else: raise ProtocolError("Unexpected message {0}".format(msg.__class__))
# noinspection PyUnusedLocal
[docs] def onClose(self, wasClean): """ Implements :func:`autobahn.wamp.interfaces.ITransportHandler.onClose` """ self._transport = None if self._session_id: ## fire callback and close the transport try: self.onLeave(types.CloseDetails()) except Exception as e: if self.debug: print("exception raised in onLeave callback: {0}".format(e)) self._session_id = None self.onDisconnect()
[docs] def onChallenge(self, challenge): """ Implements :func:`autobahn.wamp.interfaces.ISession.onChallenge` """ raise Exception("received authentication challenge, but onChallenge not implemented")
[docs] def onJoin(self, details): """ Implements :func:`autobahn.wamp.interfaces.ISession.onJoin` """
[docs] def onLeave(self, details): """ Implements :func:`autobahn.wamp.interfaces.ISession.onLeave` """ self.disconnect()
[docs] def leave(self, reason = None, log_message = None): """ Implements :func:`autobahn.wamp.interfaces.ISession.leave` """ if not self._session_id: raise Exception("not joined") if not self._goodbye_sent: if not reason: reason = u"wamp.close.normal" msg = wamp.message.Goodbye(reason = reason, message = log_message) self._transport.send(msg) self._goodbye_sent = True else: raise SessionNotReady(u"Already requested to close the session")
[docs] def publish(self, topic, *args, **kwargs): """ Implements :func:`autobahn.wamp.interfaces.IPublisher.publish` """ if six.PY2 and type(topic) == str: topic = six.u(topic) assert(type(topic) == six.text_type) if not self._transport: raise exception.TransportLost() request = util.id() if 'options' in kwargs and isinstance(kwargs['options'], types.PublishOptions): opts = kwargs.pop('options') msg = message.Publish(request, topic, args = args, kwargs = kwargs, **opts.options) else: opts = None msg = message.Publish(request, topic, args = args, kwargs = kwargs) if opts and opts.options['acknowledge'] == True: d = self._create_future() self._publish_reqs[request] = d, opts self._transport.send(msg) return d else: self._transport.send(msg) return
[docs] def subscribe(self, handler, topic = None, options = None): """ Implements :func:`autobahn.wamp.interfaces.ISubscriber.subscribe` """ assert((callable(handler) and topic is not None) or hasattr(handler, '__class__')) if topic and six.PY2 and type(topic) == str: topic = six.u(topic) assert(topic is None or type(topic) == six.text_type) assert(options is None or isinstance(options, types.SubscribeOptions)) if not self._transport: raise exception.TransportLost() def _subscribe(obj, handler, topic, options): request = util.id() d = self._create_future() self._subscribe_reqs[request] = (d, obj, handler, topic, options) if options is not None: msg = message.Subscribe(request, topic, **options.options) else: msg = message.Subscribe(request, topic) self._transport.send(msg) return d if callable(handler): ## subscribe a single handler return _subscribe(None, handler, topic, options) else: ## subscribe all methods on an object decorated with "wamp.subscribe" dl = [] test = lambda x: inspect.ismethod(x) or inspect.isfunction(x) for k in inspect.getmembers(handler.__class__, test): proc = k[1] if "_wampuris" in proc.__dict__: pat = proc.__dict__["_wampuris"][0] if pat.is_handler(): uri = pat.uri() dl.append(_subscribe(handler, proc, uri, options)) return self._gather_futures(dl, consume_exceptions = True)
def _unsubscribe(self, subscription): """ Called from :meth:`autobahn.wamp.protocol.Subscription.unsubscribe` """ assert(isinstance(subscription, Subscription)) assert subscription.active assert(subscription.id in self._subscriptions) if not self._transport: raise exception.TransportLost() request = util.id() d = self._create_future() self._unsubscribe_reqs[request] = (d, subscription) msg = message.Unsubscribe(request, subscription.id) self._transport.send(msg) return d
[docs] def call(self, procedure, *args, **kwargs): """ Implements :func:`autobahn.wamp.interfaces.ICaller.call` """ if six.PY2 and type(procedure) == str: procedure = six.u(procedure) assert(isinstance(procedure, six.text_type)) if not self._transport: raise exception.TransportLost() request = util.id() if 'options' in kwargs and isinstance(kwargs['options'], types.CallOptions): opts = kwargs.pop('options') msg = message.Call(request, procedure, args = args, kwargs = kwargs, **opts.options) else: opts = None msg = message.Call(request, procedure, args = args, kwargs = kwargs) ## FIXME #def canceller(_d): # cancel_msg = message.Cancel(request) # self._transport.send(cancel_msg) #d = Deferred(canceller) d = self._create_future() self._call_reqs[request] = d, opts self._transport.send(msg) return d
[docs] def register(self, endpoint, procedure = None, options = None): """ Implements :func:`autobahn.wamp.interfaces.ICallee.register` """ assert((callable(endpoint) and procedure is not None) or hasattr(endpoint, '__class__')) if procedure and six.PY2 and type(procedure) == str: procedure = six.u(procedure) assert(procedure is None or type(procedure) == six.text_type) assert(options is None or isinstance(options, types.RegisterOptions)) if not self._transport: raise exception.TransportLost() def _register(obj, endpoint, procedure, options): request = util.id() d = self._create_future() self._register_reqs[request] = (d, obj, endpoint, procedure, options) if options is not None: msg = message.Register(request, procedure, **options.options) else: msg = message.Register(request, procedure) self._transport.send(msg) return d if callable(endpoint): ## register a single callable ## return _register(None, endpoint, procedure, options) else: ## register all methods on an object ## decorated with "wamp.register" ## dl = [] test = lambda x: inspect.ismethod(x) or inspect.isfunction(x) for k in inspect.getmembers(endpoint.__class__, test): proc = k[1] if "_wampuris" in proc.__dict__: pat = proc.__dict__["_wampuris"][0] if pat.is_endpoint(): uri = pat.uri() dl.append(_register(endpoint, proc, uri, options)) return self._gather_futures(dl, consume_exceptions = True)
def _unregister(self, registration): """ Called from :meth:`autobahn.wamp.protocol.Registration.unregister` """ assert(isinstance(registration, Registration)) assert registration.active assert(registration.id in self._registrations) if not self._transport: raise exception.TransportLost() request = util.id() d = self._create_future() self._unregister_reqs[request] = (d, registration) msg = message.Unregister(request, registration.id) self._transport.send(msg) return d
IPublisher.register(ApplicationSession) ISubscriber.register(ApplicationSession) ICaller.register(ApplicationSession) #ICallee.register(ApplicationSession) ## FIXME: ".register" collides with the ABC "register" method ITransportHandler.register(ApplicationSession)
[docs]class ApplicationSessionFactory: """ WAMP endpoint session factory. """ session = ApplicationSession """ WAMP application session class to be used in this factory. """ def __init__(self, config = None): """ :param config: The default component configuration. :type config: instance of :class:`autobahn.wamp.types.ComponentConfig` """ self.config = config or types.ComponentConfig(realm = u"default") def __call__(self): """ Creates a new WAMP application session. :returns: -- An instance of the WAMP application session class as given by `self.session`. """ session = self.session(self.config) session.factory = self return session
Reactive Manifesto: We are reactive banner