def es(self, url, query={}, body={}):
     http_client = AsyncHTTPClient()         
     if not url.startswith('/'):
         url = '/'+url
     for arg in query:
         if not isinstance(query[arg], list):
             query[arg] = [query[arg]]
     try:
         response = yield http_client.fetch(
             'http://{}{}?{}'.format(
                 config['api']['elasticsearch'],
                 url,
                 utils.url_encode_tornado_arguments(query) \
                     if query else '',
             ),
             method='POST' if body else 'GET',
             body=utils.json_dumps(body) if body else None,
         )
         return utils.json_loads(response.body)
     except HTTPError as e:
         try:
             extra = utils.json_loads(e.response.body)
         except:
             extra = {'error': e.response.body.decode('utf-8')}
         raise exceptions.Elasticsearch_exception(
             e.code,
             extra,
         )

class ExpandSearch(Spiderman):
    """Expands youtube video network.
    """

    functions = 'expand'

    def search(self):

        self.done = False
        self.requests_made = 1
        self.network = {}

        # callback function parameters
        search_key = self.search_key = str(datetime.now())
        related_search = self.related_search
        client = self.client

        cb = lambda x: related_search(x, 0)

        video_ids = [str(k) for k in self.input_object['seed']]

        global pages
        global max_results

        self.http_client = AsyncHTTPClient()

        for video_id in video_ids:
            self.http_client.fetch("http://gdata.youtube.com/feeds/api/videos/{}/related?alt=jsonc&v=2".format(video_id),
                              callback=cb)

    def test_post_participants_valid_request(self):
        db = self.get_db_client().udaan

        event = dict(
                eventName="dummyEvent",
                currentRound=0,
        )

        body = dict(
                names="Dummy Names",
                mobileNumber="9123456789"
        )

        request_body = json.dumps(body)

        event_id = yield db.events.insert(event)
        str_event_id = str(event_id)
        http_client = AsyncHTTPClient()
        response = yield http_client.fetch("http://localhost:8000/api/event_management/participants", method="POST",
                                           headers=dict(Authorization=str_event_id), body=request_body)
        response_body = json.loads(response.body.decode())
        self.assertEqual(response.code, 200)
        self.assertEqual(response_body["status"], 200)

        participant = yield db.participants.find_one({"_id": ObjectId(response_body["message"])})
        yield db.participants.remove({"_id": participant["_id"]})
        yield db.events.remove({"_id": event_id})
        self.assertEqual(response_body["message"], participant["_id"].__str__())

    def test_api(self):
        print("Running test")
        self.mock_requests = {}
        # invoke service to be tested
        client = AsyncHTTPClient(self.io_loop)

        client.fetch(self.test_data['service_endpoint'], self.stop)
        

        mocks_with_assertions = [x for x in self.test_data['mocks'] if 'body' in x['mock']['request']]
        for mock in mocks_with_assertions:
            self.wait(timeout=30)
            self.assertEqual(flatten_text(self.mock_requests[mock['mock']['name']].decode("utf-8")), flatten_text(mock['mock']['request']['body']))
            #TODO: Assert request headers
        response = self.wait()
        # print(response)

        # perform assertions
        for assertion in self.test_data['assertions']:
            if 'http_code' in assertion:
                self.assertEqual(response.code, assertion['http_code'])
            if 'response' in assertion:
                self.assertEqual(flatten_text(response.body.decode("utf-8")), flatten_text(assertion['response']))
            if 'content-type' in assertion:
                self.assertEqual(response.headers['Content-Type'], assertion['content-type'])

    def __sweep(self):
        """
        the function called by the cleanup service
        """
        #print 'candid: ' + str(len(self.__remove_candidates))
        self.__log.info('cleanup service - Sweep started')
        for sharing_secret in self.__remove_candidates:
            #cleanup all the listeners
            candidate = self.__remove_candidates[sharing_secret]
            #In case a lot of actions are waiting to be executed
            #and are clogged in the space, don't clean it up give it
            #a chance for another sweeping period
            if not candidate.is_being_processed():
                self.__log.info('cleanup service - cleaning candidate for %s' % sharing_secret)
                candidate.cleanup()
                #notify the load balancer of the cleanup
                http = AsyncHTTPClient()
                load_balancer = Properties.load_balancer_url
                url = '/'.join([load_balancer, 'SharingFactory',sharing_secret])
                http.fetch(url, method='DELETE', callback=None)
                #yield gen.Task(http.fetch, url, method = 'DELETE')
                #remove if from stored sharing spaces
                del(self.__sharing_spaces[sharing_secret])
            else:
                self.__log.info('cleanup service - skipping cleaning candidate for %s is being processed' % sharing_secret)


        #now nominate every one
        self.__remove_candidates.clear()
        for sharing_secret in self.__sharing_spaces:
            self.__remove_candidates[sharing_secret] = \
                self.__sharing_spaces[sharing_secret]
        self.__log.info('cleanup service - Sweep finished')
        self.timer = Timer(self.SWEEP_PERIOD, self.__sweep)
        self.timer.start()

class TaskPool:
    def __init__(self, maxClients):
        self._ioloop = ioloop.IOLoop()
        self._httpClient = AsyncHTTPClient(self._ioloop, maxClients)
        self._taskNum = 0
        
    def run(self):
        self._check()
        self._ioloop.start()

    def spawn(self, request, callback, **kwargs):
        def wapped(response):
            self._taskNum -= 1
            try:
                callback(response)
            except:
                print 'spwan error:', traceback.format_exc()
                pass
                
        self._taskNum += 1
        self._httpClient.fetch(request, wapped, **kwargs)

    def _check(self):
        def callback():
            if self._taskNum == 0:
                self._ioloop.stop()

            return self._check()
                
        self._ioloop.add_callback(callback)

	def access_token_for_id(cls, id, callback):
		"""Returns the access token for an id, acquiring a new one if necessary."""
		token = Cache.get(cls.auth_cache_key_template % id)
		if token:
			return IOLoop.instance().add_callback(lambda: callback(token))

		# If we don't have an access token cached, see if we have a refresh token
		token = TokenIdMapping.lookup_refresh_token(id)
		if token:
			post_body = urllib.urlencode({
				'client_id': Config.get('oauth', 'client-id'),
				'client_secret': Config.get('oauth', 'client-secret'),
				'refresh_token': token,
				'grant_type': 'refresh_token',
			})
			http_client = AsyncHTTPClient()
			return http_client.fetch(
				'https://accounts.google.com/o/oauth2/token',
				lambda response: cls.on_refresh_complete(response, id, callback),
				method='POST',
				body=post_body,
				request_timeout=20.0,
				connect_timeout=15.0,
			)
		else:
			logging.error("Unable to update access token for %s, no refresh token stored.", id)
			return IOLoop.instance().add_callback(lambda: callback(None))

def test_with_data(e, s, a, b):
    ss = HTTPScheduler(s)
    ss.listen(0)

    L = e.map(inc, [1, 2, 3])
    L2 = yield e._scatter(['Hello', 'world!'])
    yield _wait(L)

    client = AsyncHTTPClient()
    response = yield client.fetch('http://localhost:%d/memory-load.json' %
                                  ss.port)
    out = json.loads(response.body.decode())

    assert all(isinstance(v, int) for v in out.values())
    assert set(out) == {a.address, b.address}
    assert sum(out.values()) == sum(map(getsizeof,
                                        [1, 2, 3, 'Hello', 'world!']))

    response = yield client.fetch('http://localhost:%s/memory-load-by-key.json'
                                  % ss.port)
    out = json.loads(response.body.decode())
    assert set(out) == {a.address, b.address}
    assert all(isinstance(v, dict) for v in out.values())
    assert all(k in {'inc', 'data'} for d in out.values() for k in d)
    assert all(isinstance(v, int) for d in out.values() for v in d.values())

    assert sum(v for d in out.values() for v in d.values()) == \
            sum(map(getsizeof, [1, 2, 3, 'Hello', 'world!']))

    ss.stop()

def test_with_status(e, s, a, b):
    ss = HTTPScheduler(s)
    ss.listen(0)

    client = AsyncHTTPClient()
    response = yield client.fetch('http://localhost:%d/tasks.json' %
                                  ss.port)
    out = json.loads(response.body.decode())
    assert out['total'] == 0
    assert out['processing'] == 0
    assert out['failed'] == 0
    assert out['in-memory'] == 0
    assert out['ready'] == 0
    assert out['waiting'] == 0

    L = e.map(div, range(10), range(10))
    yield _wait(L)

    client = AsyncHTTPClient()
    response = yield client.fetch('http://localhost:%d/tasks.json' %
                                  ss.port)
    out = json.loads(response.body.decode())
    assert out['failed'] == 1
    assert out['in-memory'] == 9
    assert out['ready'] == 0
    assert out['total'] == 10
    assert out['waiting'] == 0

    ss.stop()

    def post(self):

        json_str = self.get_argument("json_msg")
        # print "json_str: ",json_str
        value_obj = json.loads(json_str)

        com_val = COMMAND_URL_DICT[value_obj["command"]]
        com_url = com_val[0]
        com_func = com_val[1]
        url = "http://115.28.143.67:" + str(PORT) + com_url
        print "---------------------------------------"
        print "request url: " + url
        print "request json: " + json_str
        print "---------------------------------------"

        if "GET" == com_func:
            request = HTTPRequest(url, "GET")
            http = AsyncHTTPClient()
            response = yield http.fetch(request)
            print "---------------------------------------"
            print "response json: " + response.body
            print "---------------------------------------"
            self.write(response.body)
        elif "POST" == com_func:
            request = HTTPRequest(url, "POST", body=json_str)
            http = AsyncHTTPClient()
            response = yield http.fetch(request)
            print "---------------------------------------"
            print "response json: " + response.body
            print "---------------------------------------"
            self.write(response.body)
        else:
            pass

 def post(self, *args, **kwargs):
     # TODO
     # save the messages to local database
     # file = self.request.files['image'][0]
     # file_name = file["filename"]
     # image = file['body']
     text = self.get_argument("text")
     data = {
         "from": "Mailgun Sandbox <[email protected]>",
         "to": "<[email protected]>",
         "subject": "Hello Udaan",
         "text": text,
     }
     data = urlencode(data)
     client = AsyncHTTPClient()
     headers_object = HTTPHeaders({"X-Mailgun-Variables": dumps({"X-Mailgun-Variables": {"password": "working"}})})
     request_object = HTTPRequest("https://api.mailgun.net/v3/sandbox1713f24a60034b5ab5e7fa0ca2faa9b6.mailgun.org"
                                  "/messages",
                                  method="POST",
                                  headers=headers_object,
                                  body=data,
                                  auth_username="api",
                                  auth_password="key-a0bd92feef0ccecb07f199b770449917"
                                  )
     print(request_object.headers.get_list("X-Mailgun-Variables"))
     response = yield client.fetch(request_object)
     client.close()
     print(response)
     if response.code == 200:
         msg = "email send successfully"
         self.respond(msg, response.code)
     else:
         msg = "Please try again"
         self.respond(msg, response.code)

 def w():
     http_client = AsyncHTTPClient()
     if req_cnt % 2 == 0:
         response = yield http_client.fetch("http://localhost:8889/wait/%s/%s" % (5, req_cnt))
     else:
         response = yield http_client.fetch("http://localhost:8890/wait/%s/%s" % (1, req_cnt))
     print ">>>> response >>>", response, response.body, handler, handler.req_cnt

    def tsend_log(self,
            message,
            severity,
            filename=None,
            url=None,
            status_code=None,
            headers=None,
            parameters=None,
            stacktrace=False):
        from tornado.httpclient import AsyncHTTPClient

        d = self._build_message(
                message,
                severity,
                filename,
                url,
                status_code,
                headers,
                parameters,
                stacktrace)

        # want to use the better client here.
        AsyncHTTPClient.configure("tornado.curl_httpclient.CurlAsyncHTTPClient")

        AsyncHTTPClient().fetch(
                self._build_url(),
                lambda resp: None,
                method="POST",
                body=json.dumps(d),
                headers=headers)

def get_motion_detection(camera_id, callback):
    from tornado.httpclient import HTTPRequest, AsyncHTTPClient
    
    thread_id = camera_id_to_thread_id(camera_id)
    if thread_id is None:
        error = 'could not find thread id for camera with id %s' % camera_id
        logging.error(error)
        return callback(error=error)

    url = 'http://127.0.0.1:7999/%(id)s/detection/status' % {'id': thread_id}
    
    def on_response(response):
        if response.error:
            return callback(error=utils.pretty_http_error(response))

        enabled = bool(response.body.lower().count('active'))
        
        logging.debug('motion detection is %(what)s for camera with id %(id)s' % {
                'what': ['disabled', 'enabled'][enabled],
                'id': camera_id})

        callback(enabled)

    request = HTTPRequest(url, connect_timeout=_MOTION_CONTROL_TIMEOUT, request_timeout=_MOTION_CONTROL_TIMEOUT)
    http_client = AsyncHTTPClient()
    http_client.fetch(request, callback=on_response)

def test_broadcast(s, a, b):
    ss = HTTPScheduler(s)
    ss.listen(0)
    s.services['http'] = ss

    aa = HTTPWorker(a)
    aa.listen(0)
    a.services['http'] = aa
    a.service_ports['http'] = aa.port
    s.worker_info[a.address]['services']['http'] = aa.port

    bb = HTTPWorker(b)
    bb.listen(0)
    b.services['http'] = bb
    b.service_ports['http'] = bb.port
    s.worker_info[b.address]['services']['http'] = bb.port

    client = AsyncHTTPClient()

    a_response = yield client.fetch('http://localhost:%d/info.json' % aa.port)
    b_response = yield client.fetch('http://localhost:%d/info.json' % bb.port)
    s_response = yield client.fetch('http://localhost:%d/broadcast/info.json'
                                    % ss.port)
    assert (json.loads(s_response.body.decode()) ==
            {a.address: json.loads(a_response.body.decode()),
             b.address: json.loads(b_response.body.decode())})

    ss.stop()
    aa.stop()
    bb.stop()

def set_motion_detection(camera_id, enabled):
    from tornado.httpclient import HTTPRequest, AsyncHTTPClient
    
    thread_id = camera_id_to_thread_id(camera_id)
    if thread_id is None:
        return logging.error('could not find thread id for camera with id %s' % camera_id)
    
    if not enabled:
        _motion_detected[camera_id] = False
    
    logging.debug('%(what)s motion detection for camera with id %(id)s' % {
            'what': ['disabling', 'enabling'][enabled],
            'id': camera_id})
    
    url = 'http://127.0.0.1:7999/%(id)s/detection/%(enabled)s' % {
            'id': thread_id,
            'enabled': ['pause', 'start'][enabled]}
    
    def on_response(response):
        if response.error:
            logging.error('failed to %(what)s motion detection for camera with id %(id)s: %(msg)s' % {
                    'what': ['disable', 'enable'][enabled],
                    'id': camera_id,
                    'msg': utils.pretty_http_error(response)})
        
        else:
            logging.debug('successfully %(what)s motion detection for camera with id %(id)s' % {
                    'what': ['disabled', 'enabled'][enabled],
                    'id': camera_id})

    request = HTTPRequest(url, connect_timeout=_MOTION_CONTROL_TIMEOUT, request_timeout=_MOTION_CONTROL_TIMEOUT)
    http_client = AsyncHTTPClient()
    http_client.fetch(request, on_response)

def test_unexpected_error(error, msg):
    class Handler(object):
        def handle(self, request):
            raise error

    inbound = HTTPInbound()
    inbound.start(Handler())

    client = AsyncHTTPClient()
    req = HTTPRequest(
        url='http://localhost:%s' % inbound.port,
        method='POST',
        headers={
            headers.CALLER: 'caller',
            headers.SERVICE: 'service',
            headers.PROCEDURE: 'procedure',
            headers.ENCODING: 'json',
            headers.TTL: '10000',
        },
        body='',
    )

    with pytest.raises(HTTPError) as e:
        yield client.fetch(req)

    e = e.value
    assert e.code >= 500 and e.code < 600
    assert e.response.body == msg

 def _search2(self, q, limit):
     data = []
     http_client = AsyncHTTPClient()
     requests = [HTTPRequest(url.format(quote(q)), connect_timeout=1.0, request_timeout=60.0)
                 for url, _ in self.search_interfaces2]
     responses = yield [http_client.fetch(r) for r in requests]
     i = 0
     for r in responses:
         j = json.loads(r.body.decode())
         n = self.search_interfaces2[i][1]
         i += 1
         if j['numFound'] > 0:
             docs = j['docs']
             if n == 'F10搜索':
                 stocks = j['filters'][0]['stock']
                 doclimit = limit['f10']
             else:
                 stocks = j['filters'][1]['stock']
                 if n == '研报搜索':
                     doclimit = limit['research']
                 elif n == '新闻搜索':
                     doclimit = limit['news']
                 else:
                     doclimit = limit['notice']
             data.append({
                 'name': n,
                 'data': {
                     'stock': stocks[:limit['related_stock']],
                     'docs': docs[:doclimit]
                 }
             })
     return data

def make_request1(alert_type,ticker,msg_data,reg_ids,link):
    json_data = {
        "collapse_key" : alert_type, 
        "data" : {
            "data": msg_data,
            "ticker" : ticker,
            "link" : link,
        }, 
        "registration_ids": reg_ids,
    }


    url = 'https://android.googleapis.com/gcm/send'
    myKey = "" 
    data = json.dumps(json_data)
    headers = {'Content-Type': 'application/json', 'Authorization': 'key=%s' % myKey}
    
    http_client = AsyncHTTPClient()
    yield http_client.fetch(url, handle_request, data, headers)




     make_request("Thank you Note! ~%s"%users["name"],"Thank you for letting us know!",
    "Thanks for posting %s"%(event["name"]),[androids["reg_id"]],"findergpstracking://gpstracking#/app/setup")

def run_crawler():
    """Run a crawler iteration"""
    http_client = AsyncHTTPClient()
    # request OVH availablility API asynchronously
    response = yield http_client.fetch(URL)
    response_json = json.loads(response.body.decode('utf-8'))
    if not response_json or not response_json['answer']:
        return
    availability = response_json['answer']['availability']
    for item in availability:
        # look for servers of required types in OVH availability list
        if SERVER_TYPES.get(item['reference']) in config['servers']:
            # make a flat list of zones where servers are available
            available_zones = [e['zone'] for e in item['zones']
                               if e['availability'] != 'unavailable']
            # iterate over all tacked zones and set availability states
            for zone in config['zones']:
                server = SERVER_TYPES[item['reference']]
                state_id = '%s_available_in_%s' % (server, zone)
                # update state for each tracked zone
                text = "Server %s is available in %s" % (server, zone)
                message = {
                    'title': "Server %s available" % server,
                    'text': text,
                    'url': "http://www.kimsufi.com/fr/index.xml"
                }
                update_state(state_id, zone in available_zones, message)