def test_create_future_explicit_loop(framework):
    """
    process events on alternate loop= for create_future later
    """
    pytest.importorskip('asyncio')
    if txaio.using_twisted:
        pytest.skip()

    import asyncio

    alt_loop = asyncio.new_event_loop()

    txa = txaio.with_config(loop=alt_loop)
    f = txa.create_future()

    results = []
    f.add_done_callback(lambda r: results.append(r.result()))

    assert results == []
    txaio.resolve(f, 'some result')

    # run_once() runs the txaio.config.loop so we shouldn't get any
    # results until we spin alt_loop
    assert results == []
    run_once()
    assert results == []
    with replace_loop(alt_loop):
        run_once()
    assert results == ['some result']

 def on_leave(session, details):
     self.log.info(
         "session leaving '{details.reason}'",
         details=details,
     )
     if self._entry and not txaio.is_called(done):
         txaio.resolve(done, None)

            def process(signature_raw):
                # convert the raw signature into a hex encode value (unicode string)
                signature_hex = binascii.b2a_hex(signature_raw).decode('ascii')

                # we return the concatenation of the signature and the message signed (96 bytes)
                data_hex = binascii.b2a_hex(data).decode('ascii')

                sig = signature_hex + data_hex
                txaio.resolve(d2, sig)

 def connectionLost(self, reason):
     self.log.debug("WampRawSocketProtocol: connection lost: reason = '{reason}'", reason=reason)
     txaio.resolve(self.is_closed, self)
     try:
         wasClean = isinstance(reason.value, ConnectionDone)
         self._session.onClose(wasClean)
     except Exception as e:
         # silently ignore exceptions raised here ..
         self.log.warn("WampRawSocketProtocol: ApplicationSession.onClose raised ({err})", err=e)
     self._session = None

 def methodReturnReceived(self, mret):
     """
     Called when a method return message is received
     """
     d, timeout = self._pendingCalls.get(mret.reply_serial, (None,None))
     if timeout:
         timeout.cancel()
     if d:
         del self._pendingCalls[ mret.reply_serial ]
         txaio.resolve(d, mret)

 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 _notify_some(receivers):

                                # we do a first pass over the proposed chunk of receivers
                                # because not all of them will have a transport, and if this
                                # will be the last chunk of receivers we need to figure out
                                # which event is last...
                                receivers_this_chunk = []
                                for receiver in receivers[:chunk_size]:
                                    if receiver._session_id and receiver._transport:
                                        receivers_this_chunk.append(receiver)
                                    else:
                                        vanished_receivers.append(receiver)

                                receivers = receivers[chunk_size:]

                                # XXX note there's still going to be some edge-cases here .. if
                                # we are NOT the last chunk, but all the next chunk's receivers
                                # (could be only 1 in that chunk!) vanish before we run our next
                                # batch, then a "last" event will never go out ...

                                # we now actually do the deliveries, but now we know which
                                # receiver is the last one
                                if receivers or not self._router.is_traced:

                                    # NOT the last chunk (or we're not traced so don't care)
                                    for receiver in receivers_this_chunk:

                                        # send out WAMP msg to peer
                                        self._router.send(receiver, msg)
                                        if self._event_store or storing_event:
                                            self._event_store.store_event_history(publication, subscription.id, receiver)
                                else:
                                    # last chunk, so last receiver gets the different message
                                    for receiver in receivers_this_chunk[:-1]:
                                        self._router.send(receiver, msg)
                                        if self._event_store or storing_event:
                                            self._event_store.store_event_history(publication, subscription.id, receiver)

                                    # FIXME: I don't get the following comment and code path. when, how? and what to
                                    # do about event store? => storing_event
                                    #
                                    # we might have zero valid receivers
                                    if receivers_this_chunk:
                                        self._router.send(receivers_this_chunk[-1], last_msg)
                                        # FIXME: => storing_event

                                if receivers:
                                    # still more to do ..
                                    return txaio.call_later(0, _notify_some, receivers)
                                else:
                                    # all done! resolve all_d, which represents all receivers
                                    # to a single subscription matching the event
                                    txaio.resolve(all_d, None)

def test_callback(framework):
    f = txaio.create_future()
    results = []

    def cb(f):
        results.append(f)
    txaio.add_callbacks(f, cb, None)
    txaio.resolve(f, "it worked")

    run_once()

    assert len(results) == 1
    assert results[0] == "it worked"

 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 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)

 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)

 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 test_chained_callback(framework):
    """
    Chain two callbacks where the first one alters the value.
    """
    calls = []

    def callback0(arg):
        calls.append(arg)
        return arg + " pray I do not alter it futher"

    def callback1(arg):
        calls.append(arg)

    f = txaio.create_future()
    txaio.add_callbacks(f, callback0, None)
    txaio.add_callbacks(f, callback1, None)
    txaio.resolve(f, "the deal")

    run_once()

    assert len(calls) == 2
    assert calls[0] == "the deal"
    assert calls[1] == "the deal pray I do not alter it futher"

 def onJoin(self, details):
     txaio.resolve(d, None)

from __future__ import print_function
import txaio

def cb(value):
    print("Callback:", value)
    return value  # should always return input arg

def eb(fail):
    # fail will implement txaio.IFailedPromise
    print("Errback:", fail)
    # fail.printTraceback()
    return fail  # should always return input arg

f0 = txaio.create_future()
f1 = txaio.create_future()
txaio.add_callbacks(f0, cb, eb)
txaio.add_callbacks(f1, cb, eb)

# ...

txaio.reject(f0, RuntimeError("it failed"))
# or can just "txaio.reject(f0)" if inside an except: block
txaio.resolve(f1, "The answer is: 42")

if txaio.using_asyncio:
    # for twisted, we don't need to enter the event-loop for this
    # simple example (since all results are already available), but
    # you'd simply use reactor.run()/.stop() or task.react() as normal
    import asyncio
    asyncio.get_event_loop().run_until_complete(f1)

#.........这里部分代码省略.........

                        handler = subscription.handler

                        invoke_args = (handler.obj,) if handler.obj else tuple()
                        if msg.args:
                            invoke_args = invoke_args + tuple(msg.args)

                        invoke_kwargs = msg.kwargs if msg.kwargs else dict()
                        if handler.details_arg:
                            invoke_kwargs[handler.details_arg] = types.EventDetails(publication=msg.publication, publisher=msg.publisher, topic=msg.topic)

                        def _error(e):
                            errmsg = 'While firing {0} subscribed under {1}.'.format(
                                handler.fn, msg.subscription)
                            return self._swallow_error(e, errmsg)

                        future = txaio.as_future(handler.fn, *invoke_args, **invoke_kwargs)
                        txaio.add_callbacks(future, None, _error)

                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:

                    # get and pop outstanding publish request
                    publish_request = self._publish_reqs.pop(msg.request)

                    # create a new publication object
                    publication = Publication(msg.publication)

                    # resolve deferred/future for publishing successfully
                    txaio.resolve(publish_request.on_reply, publication)
                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:

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

                    # create new handler subscription list for subscription ID if not yet tracked
                    if msg.subscription not in self._subscriptions:
                        self._subscriptions[msg.subscription] = []

                    subscription = Subscription(msg.subscription, self, request.handler)

                    # add handler to existing subscription
                    self._subscriptions[msg.subscription].append(subscription)

                    # resolve deferred/future for subscribing successfully
                    txaio.resolve(request.on_reply, subscription)
                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:

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

                    # if the subscription still exists, mark as inactive and remove ..

 def main_success(_):
     self.log.debug("main_success")
     txaio.resolve(done, None)

 def session_done(x):
     txaio.resolve(self._done_f, None)

 def foo(x):
     f = txaio.create_future()
     txaio.resolve(f, x * x)
     return f

 def onClose(self, wasClean, code, reason):
     txaio.resolve(self.factory._done, None)