def put_task(self, inputs, callback=None):
     """ return a Future of output."""
     f = Future()
     if callback is not None:
         f.add_done_callback(callback)
     self.input_queue.put((inputs, f))
     return f

    def get(self, timeout: Union[float, datetime.timedelta] = None) -> Awaitable[_T]:
        """Remove and return an item from the queue.

        Returns an awaitable which resolves once an item is available, or raises
        `tornado.util.TimeoutError` after a timeout.

        ``timeout`` may be a number denoting a time (on the same
        scale as `tornado.ioloop.IOLoop.time`, normally `time.time`), or a
        `datetime.timedelta` object for a deadline relative to the
        current time.

        .. note::

           The ``timeout`` argument of this method differs from that
           of the standard library's `queue.Queue.get`. That method
           interprets numeric values as relative timeouts; this one
           interprets them as absolute deadlines and requires
           ``timedelta`` objects for relative timeouts (consistent
           with other timeouts in Tornado).

        """
        future = Future()  # type: Future[_T]
        try:
            future.set_result(self.get_nowait())
        except QueueEmpty:
            self._getters.append(future)
            _set_timeout(future, timeout)
        return future

    def get_tweets(username):
        result_future = Future()
        """helper function to fetch 200 tweets for a user with @username
        """
        TWITTER_URL = 'https://api.twitter.com/1.1/statuses/user_timeline.json'
        '''
        curl --get 'https://api.twitter.com/1.1/statuses/user_timeline.json' --data 'count=200&screen_name=twitterapi' --header 'Authorization: OAuth oauth_consumer_key="BlXj0VRgkpUOrN3b6vTyJu8YB", oauth_nonce="9cb4b1aaa1fb1d79e0fbd9bc8b33f82a", oauth_signature="SrJxsOCzOTnudKQMr4nMQ0gDuRk%3D", oauth_signature_method="HMAC-SHA1", oauth_timestamp="1456006969", oauth_token="701166849883373568-bqVfk8vajGxWIlKDe94CRjMJtBvwdQQ", oauth_version="1.0"' --verbose
        '''
        auth = OAuth1('BlXj0VRgkpUOrN3b6vTyJu8YB', 'qzkhGeWIYVXod9umMuinHF2OFmJxiucQspX5JsA7aH8xs5t4DT',
                      '701166849883373568-bqVfk8vajGxWIlKDe94CRjMJtBvwdQQ', 'y3gx0F5fLyIQQFNDev8JtpPKpEUmyy3mMibxCcTK2kbZZ')

        data = {'count': 200,
                'screen_name': username}

        r = requests.get(url=TWITTER_URL, params=data, auth=auth)
        data = r.json()

        if 'errors' in data:
            raise Exception
        res = []
        for item in data:
            if 'retweeted_status' not in item.keys():
                res.append(item)
        result_future.set_result(res)
        return result_future

 def _create_stream(self, max_buffer_size, af, addr, source_ip=None,
                    source_port=None):
     # Always connect in plaintext; we'll convert to ssl if necessary
     # after one connection has completed.
     source_port_bind = source_port if isinstance(source_port, int) else 0
     source_ip_bind = source_ip
     if source_port_bind and not source_ip:
         # User required a specific port, but did not specify
         # a certain source IP, will bind to the default loopback.
         source_ip_bind = '::1' if af == socket.AF_INET6 else '127.0.0.1'
         # Trying to use the same address family as the requested af socket:
         # - 127.0.0.1 for IPv4
         # - ::1 for IPv6
     socket_obj = socket.socket(af)
     set_close_exec(socket_obj.fileno())
     if source_port_bind or source_ip_bind:
         # If the user requires binding also to a specific IP/port.
         try:
             socket_obj.bind((source_ip_bind, source_port_bind))
         except socket.error:
             socket_obj.close()
             # Fail loudly if unable to use the IP/port.
             raise
     try:
         stream = IOStream(socket_obj,
                           max_buffer_size=max_buffer_size)
     except socket.error as e:
         fu = Future()
         fu.set_exception(e)
         return fu
     else:
         return stream, stream.connect(addr)

 def resolve(self, host, port, family=0):
     if is_valid_ip(host):
         addresses = [host]
     else:
         # gethostbyname doesn't take callback as a kwarg
         fut = Future()
         self.channel.gethostbyname(host, family,
                                    lambda result, error: fut.set_result((result, error)))
         result, error = yield fut
         if error:
             raise IOError('C-Ares returned error %s: %s while resolving %s' %
                           (error, pycares.errno.strerror(error), host))
         addresses = result.addresses
     addrinfo = []
     for address in addresses:
         if '.' in address:
             address_family = socket.AF_INET
         elif ':' in address:
             address_family = socket.AF_INET6
         else:
             address_family = socket.AF_UNSPEC
         if family != socket.AF_UNSPEC and family != address_family:
             raise IOError('Requested socket family %d but got %d' %
                           (family, address_family))
         addrinfo.append((address_family, (address, port)))
     raise gen.Return(addrinfo)

 def queue_unbind(self, queue='', exchange=None, routing_key=None, arguments=None):
     f = Future()
     self.channel.queue_unbind(
         callback=lambda *a: f.set_result(a), queue=queue,
         exchange=exchange, routing_key=routing_key, arguments=arguments
     )
     return f

    def send_message(self, args, callback=None):

        command = args[0]

        if 'SUBSCRIBE' in command:
            raise NotImplementedError('Not yet.')

        # Do not allow the commands, affecting the execution of other commands,
        # to be used on shared connection.
        if command in ('WATCH', 'MULTI'):
            if self.is_shared():
                raise Exception('Command %s is not allowed while connection '
                                'is shared!' % command)
            if command == 'WATCH':
                self._watch.add(args[1])
            if command == 'MULTI':
                self._multi = True

        # monitor transaction state, to unlock correctly
        if command in ('EXEC', 'DISCARD', 'UNWATCH'):
            if command in ('EXEC', 'DISCARD'):
                self._multi = False
            self._watch.clear()

        self.stream.write(self.format_message(args))

        future = Future()

        if callback is not None:
            future.add_done_callback(stack_context.wrap(callback))

        self.callbacks.append(future.set_result)

        return future

 def wait(self, position):
     future = Future()
     if position != self.count():
         future.set_result(dict(position=self.count(), last_word=self.last()))
     else:
         self.waiters.add(future)
     return future

    def write(self, data):
        assert isinstance(data, bytes)
        if self._closed:
            raise StreamClosedError(real_error=self.error)

        if not data:
            if self._write_future:
                return self._write_future
            future = Future()
            future.set_result(None)
            return future

        if self._write_buffer_size:
            self._write_buffer += data
        else:
            self._write_buffer = bytearray(data)
        self._write_buffer_size += len(data)
        future = self._write_future = Future()

        if not self._connecting:
            self._handle_write()
            if self._write_buffer_size:
                if not self._state & self.io_loop.WRITE:
                    self._state = self._state | self.io_loop.WRITE
                    self.io_loop.update_handler(self.fileno(), self._state)

        return future

    def fetch(self, request, callback=None, raise_error=True, **kwargs):
        if not isinstance(request, HTTPRequest):
            request = HTTPRequest(url=request, **kwargs)

        key = self.cache.create_key(request)

        # Check and return future if there is a pending request
        pending = self.pending_requests.get(key)
        if pending:
            return pending

        response = self.cache.get_response_and_time(key)
        if response:
            response.cached = True
            if callback:
                self.io_loop.add_callback(callback, response)
            future = Future()
            future.set_result(response)
            return future

        future = orig_fetch(self, request, callback, raise_error, **kwargs)

        self.pending_requests[key] = future

        def cache_response(future):
            exc = future.exception()
            if exc is None:
                self.cache.save_response(key, future.result())

        future.add_done_callback(cache_response)
        return future

 def wait_for(self, response):
     future = Future()
     response.callback = lambda resp: future.set_result(resp.response)
     if self.running:
         return future
     else:
         return self.run(future)

    def _consume(self, task_id, callback, wait_timeout):
        conn = yield self._connection_pool.get_connection()
        timeout = None
        consumer_tag = generate_consumer_tag()
        consume_future = Future()

        def _basic_cancel():
            conn.basic_cancel(consumer_tag=consumer_tag)
            consume_future.set_result(None)

        if wait_timeout:
            def _on_timeout():
                _basic_cancel()
                callback(WaitForResultTimeoutError(wait_timeout))
            timeout = self.io_loop.add_timeout(timedelta(milliseconds=wait_timeout), _on_timeout)

        try:
            def _on_result(reply):
                if timeout:
                    self.io_loop.remove_timeout(timeout)
                _basic_cancel()
                callback(reply)

            name = self.backend.binding.routing_key
            queue_declare_future = Future()
            conn.queue_declare(lambda method_frame: queue_declare_future.set_result(None),
                               queue=name,
                               durable=self.backend.persistent)
            yield queue_declare_future
            conn.basic_consume(
                consumer_callback=lambda channel, deliver, properties, reply: _on_result(reply),
                queue=name, consumer_tag=consumer_tag, no_ack=True)
        finally:
            self._connection_pool.put_connection(conn)
            yield consume_future

 def write_headers(self, start_line, headers, chunk=None, callback=None):
     """Implements `.HTTPConnection.write_headers`."""
     if self.is_client:
         self._request_start_line = start_line
         # Client requests with a non-empty body must have either a
         # Content-Length or a Transfer-Encoding.
         self._chunking_output = (
             start_line.method in ('POST', 'PUT', 'PATCH') and
             'Content-Length' not in headers and
             'Transfer-Encoding' not in headers)
     else:
         self._response_start_line = start_line
         self._chunking_output = (
             # TODO: should this use
             # self._request_start_line.version or
             # start_line.version?
             self._request_start_line.version == 'HTTP/1.1' and
             # 304 responses have no body (not even a zero-length body), and so
             # should not have either Content-Length or Transfer-Encoding.
             # headers.
             start_line.code != 304 and
             # No need to chunk the output if a Content-Length is specified.
             'Content-Length' not in headers and
             # Applications are discouraged from touching Transfer-Encoding,
             # but if they do, leave it alone.
             'Transfer-Encoding' not in headers)
         # If a 1.0 client asked for keep-alive, add the header.
         if (self._request_start_line.version == 'HTTP/1.0' and
             (self._request_headers.get('Connection', '').lower()
              == 'keep-alive')):
             headers['Connection'] = 'Keep-Alive'
     if self._chunking_output:
         headers['Transfer-Encoding'] = 'chunked'
     if (not self.is_client and
         (self._request_start_line.method == 'HEAD' or
          start_line.code == 304)):
         self._expected_content_remaining = 0
     elif 'Content-Length' in headers:
         self._expected_content_remaining = int(headers['Content-Length'])
     else:
         self._expected_content_remaining = None
     lines = [utf8("%s %s %s" % start_line)]
     lines.extend([utf8(n) + b": " + utf8(v) for n, v in headers.get_all()])
     for line in lines:
         if b'\n' in line:
             raise ValueError('Newline in header: ' + repr(line))
     if self.stream.closed():
         self._write_future = Future()
         self._write_future.set_exception(iostream.StreamClosedError())
     else:
         if callback is not None:
             self._write_callback = stack_context.wrap(callback)
         else:
             self._write_future = Future()
         data = b"\r\n".join(lines) + b"\r\n\r\n"
         if chunk:
             data += self._format_chunk(chunk)
         self._pending_write = self.stream.write(data)
         self._pending_write.add_done_callback(self._on_write_complete)
     return self._write_future

 def apply(callback) -> 'promise.Promise':
     f = Future()
     try:
         f.set_result(callback())
     except BaseException as e:
         f.set_exception(e)
     return Promise(f)

def wechat_api_mock(client, request, *args, **kwargs):

    url = urlparse(request.url)
    path = url.path.replace('/cgi-bin/', '').replace('/', '_')
    if path.startswith('_'):
        path = path[1:]
    res_file = os.path.join(_FIXTURE_PATH, '%s.json' % path)
    content = {
        'errcode': 99999,
        'errmsg': 'can not find fixture %s' % res_file,
    }
    headers = {
        'Content-Type': 'application/json'
    }
    try:
        with open(res_file, 'rb') as f:
            content = f.read().decode('utf-8')
    except (IOError, ValueError) as e:
        content['errmsg'] = 'Loads fixture {0} failed, error: {1}'.format(
            res_file,
            e
        )
        content = json.dumps(content)

    buffer = StringIO(content)
    resp = HTTPResponse(
        request,
        200,
        headers=headers,
        buffer=buffer,
    )
    future = Future()
    future.set_result(resp)
    return future

 def update_directories(self,update_dir_list):
     res_future = Future()
     res = {}
     for dirpath in update_dir_list:
         dir_list = []
         file_list = []
         try:
             for i in common.get_dir_contents(self.current_user, dirpath):
                 if i[0].startswith('.'):
                     continue
                 if i[2]:
                     dir_list.append(tuple(list(i)+[i[0].lower()]))
                     # dir_list.append(i)
                 else:
                     file_list.append(tuple(list(i)+[i[0].lower()]))
                     # file_list.append(i)
             # dir_list.sort()
             # file_list.sort()
             dir_list = sorted(dir_list,key=operator.itemgetter(3))
             file_list = sorted(file_list,key=operator.itemgetter(3))
             res.update({dirpath:dir_list+file_list})
         except common.MissingFileError:
             continue
     res_future.set_result(res)
     return res_future

 def exchange_declare(self, exchange, exchange_type='direct', passive=False, durable=False,
                      auto_delete=False, internal=False, nowait=False, arguments=None, type=None):
     f = Future()
     self.channel.exchange_declare(lambda *a: f.set_result(a), exchange=exchange, exchange_type=exchange_type,
                                   passive=passive, durable=durable, auto_delete=auto_delete,
                                   internal=internal, nowait=nowait, arguments=arguments, type=type)
     return f

        def wrapper(*args, **kargs):
            future = Future()
            future.set_result(self._response)

            with patch.object(AsyncHTTPClient, "fetch", return_value=future):
                with patch.object(Client, "fetch", return_value=future):
                    yield coroutine(*args, **kargs)

    def test_future_etree_element(self):
        d = Doc('a')
        f = Future()
        f.set_result(etree.Element('b'))
        d.put(f)

        self.assertXmlEqual(d.to_etree_element(), """<?xml version='1.0' encoding='utf-8'?>\n<a><b/></a>""")

    def test_rank_calls_sort_and_returns_output(self):
        #setup the response from the summarum endpoint
        expected_result = [
            ('a', 'b', 10.0),
            ('c', 'd', 1.0)
        ]
        future = Future()
        future.set_result(expected_result)
        self.endpoint.fetch_and_parse = Mock(return_value=future)
        
        #setup the response return value from the sort call
        expected_ranked_facts = {
            'predicate': {},
            'objects': []
        }
        self.ranking_service.sort = Mock(return_value = expected_ranked_facts)
        
        #call the function under test
        facts = {}
        ranked_facts = yield self.ranking_service.rank(facts)

        #check that sort was called
        self.ranking_service.sort.assert_called_once_with(facts)
        #check that rank returns the output from sort
        self.assertEquals(ranked_facts, expected_ranked_facts)