def test_errback(framework):
    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    try:
        raise exception
    except:
        fail = txaio.create_failure()
    txaio.reject(f, fail)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    assert exception == errors[0].value
    assert txaio.failure_traceback(errors[0]) is not None

    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

        def transport_check(_):
            self.log.debug('Entering re-connect loop')

            if not self._can_reconnect():
                err_msg = u"Component failed: Exhausted all transport connect attempts"
                self.log.info(err_msg)
                try:
                    raise RuntimeError(err_msg)
                except RuntimeError as e:
                    txaio.reject(self._done_f, e)
                    return

            while True:

                transport = next(transport_gen)

                if transport.can_reconnect():
                    transport_candidate[0] = transport
                    break

            delay = transport.next_delay()
            self.log.debug(
                'trying transport {transport_idx} using connect delay {transport_delay}',
                transport_idx=transport.idx,
                transport_delay=delay,
            )

            self._delay_f = txaio.sleep(delay)
            txaio.add_callbacks(self._delay_f, attempt_connect, error)

def test_errback_reject_no_args():
    """
    txaio.reject() with no args
    """

    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    try:
        raise exception
    except:
        txaio.reject(f)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    assert exception == errors[0].value
    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

def test_errback_illegal_args(framework):
    '''
    non-Exception/Failures should be rejected
    '''
    f = txaio.create_future()
    try:
        txaio.reject(f, object())
        assert "should have raised exception."
    except RuntimeError:
        pass

 def lost(fail):
     rtn = orig(fail)
     if not txaio.is_called(done):
         # asyncio will call connection_lost(None) in case of
         # a transport failure, in which case we create an
         # appropriate exception
         if fail is None:
             fail = TransportLost("failed to complete connection")
         txaio.reject(done, fail)
     return rtn

 def error(fail):
     self._delay_f = None
     if self._stopping:
         # might be better to add framework-specific checks in
         # subclasses to see if this is CancelledError (for
         # Twisted) and whatever asyncio does .. but tracking
         # if we're in the shutdown path is fine too
         txaio.resolve(self._done_f, None)
     else:
         self.log.info("Internal error {msg}", msg=txaio.failure_message(fail))
         self.log.debug("{tb}", tb=txaio.failure_format_traceback(fail))
         txaio.reject(self._done_f, fail)

 def on_error(err):
     """
     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
     """
     transport.connect_failures += 1
     # something bad has happened, and maybe didn't get caught
     # upstream yet
     if not txaio.is_called(done):
         txaio.reject(done, err)

 def notify_connect_error(fail):
     chain_f = txaio.create_future()
     # hmm, if connectfailure took a _Transport instead of
     # (or in addition to?) self it could .failed() the
     # transport and we could do away with the is_fatal
     # listener?
     handler_f = self.fire('connectfailure', self, fail.value)
     txaio.add_callbacks(
         handler_f,
         lambda _: txaio.reject(chain_f, fail),
         lambda _: txaio.reject(chain_f, fail)
     )
     return chain_f

 def on_leave(session, details):
     self.log.info(
         "session leaving '{details.reason}'",
         details=details,
     )
     if not txaio.is_called(done):
         if details.reason in [u"wamp.close.normal"]:
             txaio.resolve(done, None)
         else:
             f = txaio.create_failure(
                 ApplicationError(details.reason)
             )
             txaio.reject(done, f)

                    def connect_error(fail):
                        if isinstance(fail.value, asyncio.CancelledError):
                            reconnect[0] = False
                            txaio.reject(done_f, fail)
                            return

                        self.log.debug(u'component failed: {error}', error=txaio.failure_message(fail))
                        self.log.debug(u'{tb}', tb=txaio.failure_format_traceback(fail))
                        # If this is a "fatal error" that will never work,
                        # we bail out now
                        if isinstance(fail.value, ApplicationError):
                            if fail.value.error in [u'wamp.error.no_such_realm']:
                                reconnect[0] = False
                                self.log.error(u"Fatal error, not reconnecting")
                                txaio.reject(done_f, fail)
                                return

                            self.log.error(u"{msg}", msg=fail.value.error_message())
                            return one_reconnect_loop(None)

                        elif isinstance(fail.value, OSError):
                            # failed to connect entirely, like nobody
                            # listening etc.
                            self.log.info(u"Connection failed: {msg}", msg=txaio.failure_message(fail))
                            return one_reconnect_loop(None)

                        elif _is_ssl_error(fail.value):
                            # Quoting pyOpenSSL docs: "Whenever
                            # [SSL.Error] is raised directly, it has a
                            # list of error messages from the OpenSSL
                            # error queue, where each item is a tuple
                            # (lib, function, reason). Here lib, function
                            # and reason are all strings, describing where
                            # and what the problem is. See err(3) for more
                            # information."
                            self.log.error(u"TLS failure: {reason}", reason=fail.value.args[1])
                            self.log.error(u"Marking this transport as failed")
                            transport.failed()
                        else:
                            self.log.error(
                                u'Connection failed: {error}',
                                error=txaio.failure_message(fail),
                            )
                            # some types of errors should probably have
                            # stacktraces logged immediately at error
                            # level, e.g. SyntaxError?
                            self.log.debug(u'{tb}', tb=txaio.failure_format_traceback(fail))
                            return one_reconnect_loop(None)

    def connectionLost(self, reason):
        """
        Called when the transport loses connection to the bus
        """
        if self.busName is None:
            return

        for cb in self._dcCallbacks:
            cb(self, reason)
        
        for d, timeout in self._pendingCalls.values():
            if timeout:
                timeout.cancel()
            txaio.reject(d, reason)
        self._pendingCalls = dict()

        self.objHandler.connectionLost(reason)

 def errorReceived(self, merr):
     """
     Called when an error message is received
     """
     d, timeout = self._pendingCalls.get(merr.reply_serial, (None,None))
     if timeout:
         timeout.cancel()
     if d:
         del self._pendingCalls[ merr.reply_serial ]
         e = error.RemoteError( merr.error_name )
         e.message = ''
         e.values  = []
         if merr.body:
             if isinstance(merr.body[0], six.string_types):
                 e.message = merr.body[0]
             e.values = merr.body
         txaio.reject(d, e)

def connect( reactor,  busAddress='session' ):
    """
    Connects to the specified bus and returns a
    L{twisted.internet.defer.Deferred} to the fully-connected
    L{DBusClientConnection}. 

    @param reactor: L{twisted.internet.interfaces.IReactor} implementor
    
    @param busAddress: 'session', 'system', or a valid bus address as defined by
                       the DBus specification. If 'session' (the default) or 'system'
                       is supplied, the contents of the DBUS_SESSION_BUS_ADDRESS or
                       DBUS_SYSTEM_BUS_ADDRESS environment variables will be used for
                       the bus address, respectively. If DBUS_SYSTEM_BUS_ADDRESS is not
                       set, the well-known address unix:path=/var/run/dbus/system_bus_socket
                       will be used.
    @type busAddress: C{string}

    @rtype: L{DBusClientConnection}
    @returns: Deferred to L{DBusClientConnection} 
    """
    from txdbus import endpoints

    f = DBusClientFactory()

    d = f.getConnection()

    eplist = endpoints.getDBusEndpoints(reactor, busAddress)

    eplist.reverse()

    def try_next_ep(err):
        if eplist:
            txaio.add_callbacks(eplist.pop().connect(f), None, try_next_ep)
        else:
            txaio.reject(d, error.ConnectError(string='Failed to connect to any bus address. Last error: ' + err.getErrorMessage()))

    if eplist:
        try_next_ep(None)
    else:
        txaio.reject(d, error.ConnectError(string='Failed to connect to any bus address. No valid bus addresses found'))

    return d

def test_errback_plain_exception(framework):
    '''
    reject a future with just an Exception
    '''
    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    txaio.reject(f, exception)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

def test_errback_without_except():
    '''
    Create a failure without an except block
    '''
    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    fail = txaio.create_failure(exception)
    txaio.reject(f, fail)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

        def create_session():
            cfg = ComponentConfig(self._realm, self._extra)
            try:
                self._session = session = self.session_factory(cfg)
                for auth_name, auth_config in self._authentication.items():
                    authenticator = create_authenticator(auth_name, **auth_config)
                    session.add_authenticator(authenticator)

            except Exception as e:
                # couldn't instantiate session calls, which is fatal.
                # let the reconnection logic deal with that
                f = txaio.create_failure(e)
                txaio.reject(done, f)
                raise
            else:
                # hook up the listener to the parent so we can bubble
                # up events happning on the session onto the
                # connection. This lets you do component.on('join',
                # cb) which will work just as if you called
                # session.on('join', cb) for every session created.
                session._parent = self

                # listen on leave events; if we get errors
                # (e.g. no_such_realm), an on_leave can happen without
                # an on_join before
                def on_leave(session, details):
                    self.log.info(
                        "session leaving '{details.reason}'",
                        details=details,
                    )
                    if not txaio.is_called(done):
                        if details.reason in [u"wamp.error.no_auth_method"]:
                            txaio.resolve(done, txaio.create_failure(
                                ApplicationError(
                                    u"wamp.error.no_auth_method"
                                )
                            ))
                        else:
                            txaio.resolve(done, None)
                session.on('leave', on_leave)

                # if we were given a "main" procedure, we run through
                # it completely (i.e. until its Deferred fires) and
                # then disconnect this session
                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)
                if self._entry is not None:
                    session.on('join', on_join)

                # listen on disconnect events. Note that in case we
                # had a "main" procedure, we could have already
                # resolve()'d our "done" future
                def on_disconnect(session, was_clean):
                    self.log.debug(
                        "session on_disconnect: was_clean={was_clean}",
                        was_clean=was_clean,
                    )
                    if not txaio.is_called(done):
                        if not was_clean:
                            self.log.warn(
                                u"Session disconnected uncleanly"
                            )
                        # eg the session has left the realm, and the transport was properly
                        # shut down. successfully finish the connection
                        txaio.resolve(done, None)
                session.on('disconnect', on_disconnect)

                # return the fresh session object
                return session

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

 def try_next_ep(err):
     if eplist:
         txaio.add_callbacks(eplist.pop().connect(f), None, try_next_ep)
     else:
         txaio.reject(d, error.ConnectError(string='Failed to connect to any bus address. Last error: ' + err.getErrorMessage()))

 def lost(fail):
     rtn = orig(fail)
     if not txaio.is_called(done):
         txaio.reject(done, fail)
     return rtn

#.........这里部分代码省略.........
                                # the application-level payload returned from the invoked procedure can't be serialized
                                reply = message.Error(message.Invocation.MESSAGE_TYPE, msg.request, ApplicationError.INVALID_PAYLOAD,
                                                      args=[u'error return value from invoked procedure "{0}" could not be serialized: {1}'.format(registration.procedure, e)])
                                self._transport.send(reply)
                            # we have handled the error, so we eat it
                            return None

                        self._invocations[msg.request] = InvocationRequest(msg.request, on_reply)

                        txaio.add_callbacks(on_reply, 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 as e:
                        # XXX can .cancel() return a Deferred/Future?
                        try:
                            self.onUserError(e, "While cancelling call.")
                        except:
                            pass
                    finally:
                        del self._invocations[msg.request]

            elif isinstance(msg, message.Registered):

                if msg.request in self._register_reqs:

                    # get and pop outstanding register request
                    request = self._register_reqs.pop(msg.request)

                    # create new registration if not yet tracked
                    if msg.registration not in self._registrations:
                        registration = Registration(self, msg.registration, request.procedure, request.endpoint)
                        self._registrations[msg.registration] = registration
                    else:
                        raise ProtocolError("REGISTERED received for already existing registration ID {0}".format(msg.registration))

                    txaio.resolve(request.on_reply, registration)
                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:

                    # get and pop outstanding subscribe request
                    request = self._unregister_reqs.pop(msg.request)

                    # if the registration still exists, mark as inactive and remove ..
                    if request.registration_id in self._registrations:
                        self._registrations[request.registration_id].active = False
                        del self._registrations[request.registration_id]

                    # resolve deferred/future for unregistering successfully
                    txaio.resolve(request.on_reply)
                else:
                    raise ProtocolError("UNREGISTERED received for non-pending request ID {0}".format(msg.request))

            elif isinstance(msg, message.Error):

                # remove outstanding request and get the reply deferred/future
                on_reply = None

                # ERROR reply to CALL
                if msg.request_type == message.Call.MESSAGE_TYPE and msg.request in self._call_reqs:
                    on_reply = self._call_reqs.pop(msg.request).on_reply

                # ERROR reply to PUBLISH
                elif msg.request_type == message.Publish.MESSAGE_TYPE and msg.request in self._publish_reqs:
                    on_reply = self._publish_reqs.pop(msg.request).on_reply

                # ERROR reply to SUBSCRIBE
                elif msg.request_type == message.Subscribe.MESSAGE_TYPE and msg.request in self._subscribe_reqs:
                    on_reply = self._subscribe_reqs.pop(msg.request).on_reply

                # ERROR reply to UNSUBSCRIBE
                elif msg.request_type == message.Unsubscribe.MESSAGE_TYPE and msg.request in self._unsubscribe_reqs:
                    on_reply = self._unsubscribe_reqs.pop(msg.request).on_reply

                # ERROR reply to REGISTER
                elif msg.request_type == message.Register.MESSAGE_TYPE and msg.request in self._register_reqs:
                    on_reply = self._register_reqs.pop(msg.request).on_reply

                # ERROR reply to UNREGISTER
                elif msg.request_type == message.Unregister.MESSAGE_TYPE and msg.request in self._unregister_reqs:
                    on_reply = self._unregister_reqs.pop(msg.request).on_reply

                if on_reply:
                    txaio.reject(on_reply, 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))

            else:

                raise ProtocolError("Unexpected message {0}".format(msg.__class__))