How to use the txaio.create_future function in txaio

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 test_is_future_generic(framework):
    '''
    Returning an immediate value from as_future
    '''
    f = txaio.create_future('result')

    assert txaio.is_future(f)

def connectionMade(self):
        self.log.debug('{klass}.connectionMade()', klass=self.__class__.__name__)

        # the peer we are connected to
        #
        try:
            peer = self.transport.getPeer()
        except AttributeError:
            # ProcessProtocols lack getPeer()
            self.peer = "?"
        else:
            self.peer = peer2str(peer)

        # a Future/Deferred that fires when we hit STATE_CLOSED
        self.is_closed = txaio.create_future()

        # this will hold an ApplicationSession object
        # once the RawSocket opening handshake has been
        # completed
        #
        self._session = None

        # Will hold the negotiated serializer once the opening handshake is complete
        #
        self._serializer = None

        # Will be set to True once the opening handshake is complete
        #
        self._handshake_complete = False

        # Buffer for opening handshake received bytes.

def _register(obj, fn, procedure, options):
            request_id = self._request_id_gen.next()
            on_reply = txaio.create_future()
            endpoint_obj = Endpoint(fn, obj, options.details_arg if options else None)
            if prefix is not None:
                procedure = u"{}{}".format(prefix, procedure)
            self._register_reqs[request_id] = RegisterRequest(request_id, on_reply, procedure, endpoint_obj)

            if options:
                msg = message.Register(request_id, procedure, **options.message_attr())
            else:
                msg = message.Register(request_id, procedure)

            if options:
                if options.correlation_id is not None:
                    msg.correlation_id = options.correlation_id
                if options.correlation_uri is not None:
                    msg.correlation_uri = options.correlation_uri
                if options.correlation_is_anchor is not None:

def _register(obj, fn, procedure, options):
            request_id = self._request_id_gen.next()
            on_reply = txaio.create_future()
            endpoint_obj = Endpoint(fn, obj, options.details_arg if options else None)
            self._register_reqs[request_id] = RegisterRequest(request_id, on_reply, procedure, endpoint_obj)

            if options:
                msg = message.Register(request_id, procedure, **options.message_attr())
            else:
                msg = message.Register(request_id, procedure)

            self._transport.send(msg)
            return on_reply

else:
            if options:
                msg = message.Publish(request_id,
                                      topic,
                                      args=args,
                                      kwargs=kwargs,
                                      **options.message_attr())
            else:
                msg = message.Publish(request_id,
                                      topic,
                                      args=args,
                                      kwargs=kwargs)

        if options and options.acknowledge:
            # only acknowledged publications expect a reply ..
            on_reply = txaio.create_future()
            self._publish_reqs[request_id] = PublishRequest(request_id, on_reply, was_encrypted=(encoded_payload is not None))
        else:
            on_reply = None

        try:
            # Notes:
            #
            # * this might raise autobahn.wamp.exception.SerializationError
            #   when the user payload cannot be serialized
            # * we have to setup a PublishRequest() in _publish_reqs _before_
            #   calling transpor.send(), because a mock- or side-by-side transport
            #   will immediately lead on an incoming WAMP message in onMessage()
            #
            self._transport.send(msg)
        except Exception as e:
            if request_id in self._publish_reqs:

def __init__(self):
        self.service = None
        self.is_closed = txaio.create_future()

def connection_made(self, transport):
        self.transport = transport
        peer = transport.get_extra_info('peername')
        self.peer = peer2str(peer)
        self.log.debug('RawSocker Asyncio: Connection made with peer {peer}', peer=self.peer)
        self._buffer = b''
        self._header = None
        self._wait_closed = txaio.create_future()

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, []):
            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)