def onClose(self, wasClean):
        """
        Implements :func:`autobahn.wamp.interfaces.ITransportHandler.onClose`
        """
        self._transport = None

        if self._session_id:
            # fire callback and close the transport
            d = txaio.as_future(
                self.onLeave,
                types.CloseDetails(
                    reason=types.CloseDetails.REASON_TRANSPORT_LOST,
                    message="WAMP transport was lost without closing the session before",
                ),
            )

            def _error(e):
                return self._swallow_error(e, "While firing onLeave")

            txaio.add_callbacks(d, None, _error)

            self._session_id = None

        d = txaio.as_future(self.onDisconnect, wasClean)

        def _error(e):
            return self._swallow_error(e, "While firing onDisconnect")

        txaio.add_callbacks(d, None, _error)

def test_explicit_reactor_coroutine(framework):
    """
    If we set an event-loop, Futures + Tasks should use it.
    """
    pytest.importorskip('asyncio')
    if txaio.using_twisted:
        pytest.skip()

    from asyncio import coroutine

    @coroutine
    def some_coroutine():
        yield 'nothing'

    with patch.object(txaio.config, 'loop') as fake_loop:
        txaio.as_future(some_coroutine)

        if sys.version_info < (3, 4, 2):
            assert len(fake_loop.method_calls) == 2
            c = fake_loop.method_calls[1]
            assert c[0] == 'call_soon'
        else:
            assert len(fake_loop.method_calls) == 1
            c = fake_loop.method_calls[0]
            assert c[0] == 'create_task'

def test_as_future_recursive(framework):
    '''
    Returns another Future from as_future
    '''
    errors = []
    results = []
    calls = []
    f1 = txaio.create_future_success(42)

    def method(*args, **kw):
        calls.append((args, kw))
        return f1
    f0 = txaio.as_future(method, 1, 2, 3, key='word')

    def cb(x):
        results.append(x)

    def errback(f):
        errors.append(f)

    txaio.add_callbacks(f0, cb, errback)

    run_once()

    assert len(results) == 1
    assert len(errors) == 0
    assert results[0] == 42
    assert calls[0] == ((1, 2, 3), dict(key='word'))

def test_as_future_exception(framework):
    '''
    Raises an exception from as_future
    '''
    errors = []
    results = []
    calls = []
    exception = RuntimeError("sadness")

    def method(*args, **kw):
        calls.append((args, kw))
        raise exception
    f = txaio.as_future(method, 1, 2, 3, key='word')

    def cb(x):
        results.append(x)

    def errback(f):
        errors.append(f)

    txaio.add_callbacks(f, cb, errback)

    run_once()

    assert len(results) == 0
    assert len(errors) == 1
    assert errors[0].value == exception
    assert calls[0] == ((1, 2, 3), dict(key='word'))

def test_as_future_immediate_none(framework):
    '''
    Returning None immediately from as_future
    '''
    errors = []
    results = []
    calls = []

    def method(*args, **kw):
        calls.append((args, kw))
        return None
    f = txaio.as_future(method, 1, 2, 3, key='word')

    def cb(x):
        results.append(x)

    def errback(f):
        errors.append(f)

    txaio.add_callbacks(f, cb, errback)

    run_once()

    assert len(results) == 1
    assert len(errors) == 0
    assert results[0] is None
    assert calls[0] == ((1, 2, 3), dict(key='word'))

 def fire(self, event, *args, **kwargs):
     res = []
     if event in self._listeners:
         for handler in self._listeners[event]:
             value = txaio.as_future(handler, *args, **kwargs)
             res.append(value)
     if self._parent is not None:
         res.append(self._parent.fire(event, *args, **kwargs))
     return txaio.gather(res)

 def component_start(comp):
     # the future from start() errbacks if we fail, or callbacks
     # when the component is considered "done" (so maybe never)
     d = txaio.as_future(comp.start, reactor)
     txaio.add_callbacks(
         d,
         partial(component_success, comp),
         partial(component_failure, comp),
     )
     return d

    def onOpen(self, transport):
        """
        Implements :func:`autobahn.wamp.interfaces.ITransportHandler.onOpen`
        """
        self._transport = transport
        d = txaio.as_future(self.onConnect)

        def _error(e):
            return self._swallow_error(e, "While firing onConnect")
        txaio.add_callbacks(d, None, _error)

                    def on_join(session, details):
                        self.log.debug("session on_join: {details}", details=details)
                        d = txaio.as_future(self._entry, reactor, session)

                        def setup_success(_):
                            self.log.debug("setup_success")

                        def setup_error(err):
                            self.log.debug("setup_error", err)

                        txaio.add_callbacks(d, setup_success, setup_error)

def test_gather_two():
    '''
    Wait for two Futures.
    '''

    errors = []
    results = []
    calls = []

    def foo():
        def codependant(*args, **kw):
            calls.append((args, kw))
            return 42
        return txaio.as_future(codependant)

    def method(*args, **kw):
        calls.append((args, kw))
        return "OHAI"
    f0 = txaio.as_future(method, 1, 2, 3, key='word')
    f1 = txaio.as_future(foo)

    f2 = txaio.gather([f0, f1])

    def done(arg):
        results.append(arg)

    def error(fail):
        errors.append(fail)
        # fail.printTraceback()
    txaio.add_callbacks(f2, done, error)

    await(f0)
    await(f1)
    await(f2)

    assert len(results) == 1
    assert len(errors) == 0
    assert results[0] == ['OHAI', 42] or results[0] == [42, 'OHAI']
    assert len(calls) == 2
    assert calls[0] == ((1, 2, 3), dict(key='word'))
    assert calls[1] == (tuple(), dict())

    def authorize(self, session, uri, action, options):
        """
        Authorizes a session for an action on an URI.

        Implements :func:`autobahn.wamp.interfaces.IRouter.authorize`
        """
        assert(type(uri) == str)
        assert(action in [u'call', u'register', u'publish', u'subscribe'])

        # the role under which the session that wishes to perform the given action on
        # the given URI was authenticated under
        role = session._authrole

        if role in self._roles:
            # the authorizer procedure of the role which we will call ..
            authorize = self._roles[role].authorize
            d = txaio.as_future(authorize, session, uri, action, options)
        else:
            # normally, the role should exist on the router (and hence we should not arrive
            # here), but the role might have been dynamically removed - and anyway, safety first!
            d = txaio.create_future_success(False)

        # XXX would be nicer for dynamic-authorizer authors if we
        # sanity-checked the return-value ('authorization') here
        # (i.e. is it a dict? does it have 'allow' in it? does it have
        # disallowed keys in it?)

        def got_authorization(authorization):
            # backward compatibility
            if isinstance(authorization, bool):
                authorization = {
                    u'allow': authorization,
                    u'cache': False
                }
                if action in [u'call', u'publish']:
                    authorization[u'disclose'] = False

            auto_disclose_trusted = False
            if auto_disclose_trusted and role == u'trusted' and action in [u'call', u'publish']:
                authorization[u'disclose'] = True

            self.log.debug("Authorized action '{action}' for URI '{uri}' by session {session_id} with authid '{authid}' and authrole '{authrole}' -> authorization: {authorization}",
                           session_id=session._session_id,
                           uri=uri,
                           action=action,
                           authid=session._authid,
                           authrole=session._authrole,
                           authorization=authorization)

            return authorization

        d.addCallback(got_authorization)
        return d

                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
                    details = types.CloseDetails(reply.reason, reply.message)
                    d = txaio.as_future(self.onLeave, details)

                    def _error(e):
                        return self._swallow_error(e, "While firing onLeave")
                    txaio.add_callbacks(d, None, _error)
                    # switching to the callback chain, effectively
                    # cancelling error (which we've now handled)
                    return d

 def stop(self):
     self._stopping = True
     if self._session and self._session.is_attached():
         return self._session.leave()
     elif self._delay_f:
         # This cancel request will actually call the "error" callback of
         # the _delay_f future. Nothing to worry about.
         return txaio.as_future(txaio.cancel, self._delay_f)
     # if (for some reason -- should we log warning here to figure
     # out if this can evern happen?) we've not fired _done_f, we
     # do that now (causing our "main" to exit, and thus react() to
     # quit)
     if not txaio.is_called(self._done_f):
         txaio.resolve(self._done_f, None)
     return txaio.create_future_success(None)

    def authorize(self, session, uri, action):
        """
        Authorizes a session for an action on an URI.

        Implements :func:`autobahn.wamp.interfaces.IRouter.authorize`
        """
        assert(type(uri) == six.text_type)
        assert(action in [u'call', u'register', u'publish', u'subscribe'])

        # the role under which the session that wishes to perform the given action on
        # the given URI was authenticated under
        role = session._authrole

        if role in self._roles:
            # the authorizer procedure of the role which we will call ..
            authorize = self._roles[role].authorize
            d = txaio.as_future(authorize, session, uri, action)
        else:
            # normally, the role should exist on the router (and hence we should not arrive
            # here), but the role might have been dynamically removed - and anyway, safety first!
            d = txaio.create_future_success(False)

        def got_authorization(authorization):
            # backward compatibility
            if type(authorization) == bool:
                authorization = {
                    u'allow': authorization,
                    u'cache': False
                }
                if action in [u'call', u'publish']:
                    authorization[u'disclose'] = False

            self.log.debug("Authorized action '{action}' for URI '{uri}' by session {session_id} with authid '{authid}' and authrole '{authrole}' -> authorization: {authorization}",
                           session_id=session._session_id,
                           uri=uri,
                           action=action,
                           authid=session._authid,
                           authrole=session._authrole,
                           authorization=authorization)

            return authorization

        d.addCallback(got_authorization)
        return d

                def on_join(session, details):
                    transport.connect_sucesses += 1
                    self.log.debug("session on_join: {details}", details=details)
                    d = txaio.as_future(self._entry, reactor, session)

                    def main_success(_):
                        self.log.debug("main_success")

                        def leave():
                            try:
                                session.leave()
                            except SessionNotReady:
                                # someone may have already called
                                # leave()
                                pass
                        txaio.call_later(0, leave)

                    def main_error(err):
                        self.log.debug("main_error: {err}", err=err)
                        txaio.reject(done, err)
                        session.disconnect()
                    txaio.add_callbacks(d, main_success, main_error)

    def fire(self, event, *args, **kwargs):
        """
        Fire a particular event.

        :param event: the event to fire. All other args and kwargs are
            passed on to the handler(s) for the event.

        :return: a Deferred/Future gathering all async results from
            all handlers and/or parent handlers.
        """
        # print("firing '{}' from '{}'".format(event, hash(self)))
        if self._listeners is None:
            return txaio.create_future(result=[])

        self._check_event(event)
        res = []
        for handler in self._listeners.get(event, set()):
            future = txaio.as_future(handler, *args, **kwargs)
            res.append(future)
        if self._parent is not None:
            res.append(self._parent.fire(event, *args, **kwargs))
        return txaio.gather(res, consume_exceptions=False)

def test_as_future_coroutine(framework):
    '''
    call a coroutine (asyncio)
    '''
    pytest.importorskip('asyncio')
    # can import asyncio on python3.4, but might still be using
    # twisted
    if not txaio.using_asyncio:
        return

    errors = []
    results = []
    calls = []

    from asyncio import coroutine

    @coroutine
    def method(*args, **kw):
        calls.append((args, kw))
        return 42
    f = txaio.as_future(method, 1, 2, 3, key='word')

    def cb(x):
        results.append(x)

    def errback(f):
        errors.append(f)

    txaio.add_callbacks(f, cb, errback)

    run_once()
    run_once()

    assert len(results) == 1
    assert len(errors) == 0
    assert results[0] == 42
    assert calls[0] == ((1, 2, 3), dict(key='word'))

    def processCall(self, session, call):
        """
        Implements :func:`crossbar.router.interfaces.IDealer.processCall`
        """
        # check procedure URI: for CALL, must be valid URI (either strict or loose), and
        # all URI components must be non-empty
        if self._option_uri_strict:
            uri_is_valid = _URI_PAT_STRICT_NON_EMPTY.match(call.procedure)
        else:
            uri_is_valid = _URI_PAT_LOOSE_NON_EMPTY.match(call.procedure)

        if not uri_is_valid:
            reply = message.Error(message.Call.MESSAGE_TYPE, call.request, ApplicationError.INVALID_URI, [u"call with invalid procedure URI '{0}' (URI strict checking {1})".format(call.procedure, self._option_uri_strict)])
            self._router.send(session, reply)
            return

        # get registrations active on the procedure called
        #
        registration = self._registration_map.best_matching_observation(call.procedure)

        if registration:

            # validate payload (skip in "payload_transparency" mode)
            #
            if call.payload is None:
                try:
                    self._router.validate('call', call.procedure, call.args, call.kwargs)
                except Exception as e:
                    reply = message.Error(message.Call.MESSAGE_TYPE, call.request, ApplicationError.INVALID_ARGUMENT, [u"call of procedure '{0}' with invalid application payload: {1}".format(call.procedure, e)])
                    self._router.send(session, reply)
                    return

            # authorize CALL action
            #
            d = txaio.as_future(self._router.authorize, session, call.procedure, RouterAction.ACTION_CALL)

            def on_authorize_success(authorized):

                # the call to authorize the action _itself_ succeeded. now go on depending on whether
                # the action was actually authorized or not ..
                #
                if not authorized:
                    reply = message.Error(message.Call.MESSAGE_TYPE, call.request, ApplicationError.NOT_AUTHORIZED, [u"session is not authorized to call procedure '{0}'".format(call.procedure)])
                    self._router.send(session, reply)

                else:

                    # determine callee according to invocation policy
                    #
                    if registration.extra.invoke == message.Register.INVOKE_SINGLE:
                        callee = registration.observers[0]

                    elif registration.extra.invoke == message.Register.INVOKE_FIRST:
                        callee = registration.observers[0]

                    elif registration.extra.invoke == message.Register.INVOKE_LAST:
                        callee = registration.observers[len(registration.observers) - 1]

                    elif registration.extra.invoke == message.Register.INVOKE_ROUNDROBIN:
                        callee = registration.observers[registration.extra.roundrobin_current % len(registration.observers)]
                        registration.extra.roundrobin_current += 1

                    elif registration.extra.invoke == message.Register.INVOKE_RANDOM:
                        callee = registration.observers[random.randint(0, len(registration.observers) - 1)]

                    else:
                        # should not arrive here
                        raise Exception(u"logic error")

                    # new ID for the invocation
                    #
                    invocation_request_id = self._request_id_gen.next()

                    # caller disclosure
                    #
                    if call.disclose_me:
                        caller = session._session_id
                    else:
                        caller = None

                    # for pattern-based registrations, the INVOCATION must contain
                    # the actual procedure being called
                    #
                    if registration.match != message.Register.MATCH_EXACT:
                        procedure = call.procedure
                    else:
                        procedure = None

                    if call.payload:
                        invocation = message.Invocation(invocation_request_id,
                                                        registration.id,
                                                        payload=call.payload,
                                                        timeout=call.timeout,
                                                        receive_progress=call.receive_progress,
                                                        caller=caller,
                                                        procedure=procedure,
                                                        enc_algo=call.enc_algo,
                                                        enc_key=call.enc_key,
                                                        enc_serializer=call.enc_serializer)
                    else:
#.........这里部分代码省略.........

    def processSubscribe(self, session, subscribe):
        """
        Implements :func:`crossbar.router.interfaces.IBroker.processSubscribe`
        """
        # check topic URI: for SUBSCRIBE, must be valid URI (either strict or loose), and all
        # URI components must be non-empty other than for wildcard subscriptions
        #
        if self._option_uri_strict:
            if subscribe.match == u"wildcard":
                uri_is_valid = _URI_PAT_STRICT_EMPTY.match(subscribe.topic)
            else:
                uri_is_valid = _URI_PAT_STRICT_NON_EMPTY.match(subscribe.topic)
        else:
            if subscribe.match == u"wildcard":
                uri_is_valid = _URI_PAT_LOOSE_EMPTY.match(subscribe.topic)
            else:
                uri_is_valid = _URI_PAT_LOOSE_NON_EMPTY.match(subscribe.topic)

        if not uri_is_valid:
            reply = message.Error(message.Subscribe.MESSAGE_TYPE, subscribe.request, ApplicationError.INVALID_URI, [u"subscribe for invalid topic URI '{0}'".format(subscribe.topic)])
            session._transport.send(reply)
            return

        # authorize action
        #
        d = txaio.as_future(self._router.authorize, session, subscribe.topic, RouterAction.ACTION_SUBSCRIBE)

        def on_authorize_success(authorized):
            if not authorized:
                # error reply since session is not authorized to subscribe
                #
                reply = message.Error(message.Subscribe.MESSAGE_TYPE, subscribe.request, ApplicationError.NOT_AUTHORIZED, [u"session is not authorized to subscribe to topic '{0}'".format(subscribe.topic)])

            else:
                # ok, session authorized to subscribe. now get the subscription
                #
                subscription, was_already_subscribed, is_first_subscriber = self._subscription_map.add_observer(session, subscribe.topic, subscribe.match)

                if not was_already_subscribed:
                    self._session_to_subscriptions[session].add(subscription)

                # publish WAMP meta events
                #
                if self._router._realm:
                    service_session = self._router._realm.session
                    if service_session and not subscription.uri.startswith(u'wamp.'):
                        if is_first_subscriber:
                            subscription_details = {
                                'id': subscription.id,
                                'created': subscription.created,
                                'uri': subscription.uri,
                                'match': subscription.match,
                            }
                            service_session.publish(u'wamp.subscription.on_create', session._session_id, subscription_details)
                        if not was_already_subscribed:
                            service_session.publish(u'wamp.subscription.on_subscribe', session._session_id, subscription.id)

                # acknowledge subscribe with subscription ID
                #
                reply = message.Subscribed(subscribe.request, subscription.id)

            # send out reply to subscribe requestor
            #
            session._transport.send(reply)

        def on_authorize_error(err):
            """
            the call to authorize the action _itself_ failed (note this is
            different from the call to authorize succeed, but the
            authorization being denied)
            """
            # XXX same as another code-block, can we collapse?
            self.log.failure(err)
            reply = message.Error(
                message.Subscribe.MESSAGE_TYPE,
                subscribe.request,
                ApplicationError.AUTHORIZATION_FAILED,
                [u"failed to authorize session for subscribing to topic URI '{0}': {1}".format(subscribe.topic, err.value)]
            )
            session._transport.send(reply)

        txaio.add_callbacks(d, on_authorize_success, on_authorize_error)

    def processRegister(self, session, register):
        """
        Implements :func:`crossbar.router.interfaces.IDealer.processRegister`
        """
        # check topic URI: for SUBSCRIBE, must be valid URI (either strict or loose), and all
        # URI components must be non-empty other than for wildcard subscriptions
        #
        if self._option_uri_strict:
            if register.match == u"wildcard":
                uri_is_valid = _URI_PAT_STRICT_EMPTY.match(register.procedure)
            else:
                uri_is_valid = _URI_PAT_STRICT_NON_EMPTY.match(register.procedure)
        else:
            if register.match == u"wildcard":
                uri_is_valid = _URI_PAT_LOOSE_EMPTY.match(register.procedure)
            else:
                uri_is_valid = _URI_PAT_LOOSE_NON_EMPTY.match(register.procedure)

        if not uri_is_valid:
            reply = message.Error(message.Register.MESSAGE_TYPE, register.request, ApplicationError.INVALID_URI, [u"register for invalid procedure URI '{0}' (URI strict checking {1})".format(register.procedure, self._option_uri_strict)])
            self._router.send(session, reply)
            return

        # disallow registration of procedures starting with "wamp." and  "crossbar." other than for
        # trusted sessions (that are sessions built into Crossbar.io)
        #
        if session._authrole is not None and session._authrole != u"trusted":
            is_restricted = register.procedure.startswith(u"wamp.") or register.procedure.startswith(u"crossbar.")
            if is_restricted:
                reply = message.Error(message.Register.MESSAGE_TYPE, register.request, ApplicationError.INVALID_URI, [u"register for restricted procedure URI '{0}')".format(register.procedure)])
                self._router.send(session, reply)
                return

        # get existing registration for procedure / matching strategy - if any
        #
        registration = self._registration_map.get_observation(register.procedure, register.match)
        if registration:

            # there is an existing registration, and that has an invocation strategy that only allows a single callee
            # on a the given registration
            #
            if registration.extra.invoke == message.Register.INVOKE_SINGLE:
                reply = message.Error(message.Register.MESSAGE_TYPE, register.request, ApplicationError.PROCEDURE_ALREADY_EXISTS, [u"register for already registered procedure '{0}'".format(register.procedure)])
                self._router.send(session, reply)
                return

            # there is an existing registration, and that has an invokation strategy different from the one
            # requested by the new callee
            #
            if registration.extra.invoke != register.invoke:
                reply = message.Error(message.Register.MESSAGE_TYPE, register.request, ApplicationError.PROCEDURE_EXISTS_INVOCATION_POLICY_CONFLICT, [u"register for already registered procedure '{0}' with conflicting invocation policy (has {1} and {2} was requested)".format(register.procedure, registration.extra.invoke, register.invoke)])
                self._router.send(session, reply)
                return

        # authorize action
        #
        d = txaio.as_future(self._router.authorize, session, register.procedure, RouterAction.ACTION_REGISTER)

        def on_authorize_success(authorized):
            if not authorized:
                # error reply since session is not authorized to register
                #
                reply = message.Error(message.Register.MESSAGE_TYPE, register.request, ApplicationError.NOT_AUTHORIZED, [u"session is not authorized to register procedure '{0}'".format(register.procedure)])

            else:
                # ok, session authorized to register. now get the registration
                #
                registration_extra = RegistrationExtra(register.invoke)
                registration, was_already_registered, is_first_callee = self._registration_map.add_observer(session, register.procedure, register.match, registration_extra)

                if not was_already_registered:
                    self._session_to_registrations[session].add(registration)

                # publish WAMP meta events
                #
                if self._router._realm:
                    service_session = self._router._realm.session
                    if service_session and not registration.uri.startswith(u'wamp.'):
                        if is_first_callee:
                            registration_details = {
                                'id': registration.id,
                                'created': registration.created,
                                'uri': registration.uri,
                                'match': registration.match,
                                'invoke': registration.extra.invoke,
                            }
                            service_session.publish(u'wamp.registration.on_create', session._session_id, registration_details)
                        if not was_already_registered:
                            service_session.publish(u'wamp.registration.on_register', session._session_id, registration.id)

                # acknowledge register with registration ID
                #
                reply = message.Registered(register.request, registration.id)

            # send out reply to register requestor
            #
            self._router.send(session, reply)

        def on_authorize_error(err):
            """
#.........这里部分代码省略.........