nsq_data.py

import tornado.httpclient
import logging
import json
import urllib
import functools


_http_client = None
def http_client():
    global _http_client
    if not _http_client:
        _http_client = tornado.httpclient.AsyncHTTPClient()
    return _http_client

def _process_response(response, endpoint, user_callback):
    if response.error:
        logging.warning("[%s] http error %s", endpoint, response.error)
        return user_callback(None)
    
    try:
        data = json.loads(response.body)
    except json.JSONDecodeError:
        logging.warning("[%s] failed to parse JSON: %r", endpoint, response.body)
        return user_callback(None)
    
    if data.get('status_code') != 200:
        logging.warning("[%s] endpoint responded with %d", endpoint, data.get('status_code'))
        return user_callback(None)
    
    user_callback(data.get('data', {}))

def get_topics_in_lookupd(endpoint, user_callback):
    lookupd_url = endpoint + "/topics"
    req = tornado.httpclient.HTTPRequest(lookupd_url, method="GET", 
        connect_timeout=2, request_timeout=5)
    callback = functools.partial(_process_response, endpoint=endpoint, user_callback=user_callback)
    http_client().fetch(req, callback=callback)

def get_producers_for_topic(endpoint, topic, user_callback):
    lookupd_url = endpoint + "/lookup?topic=" + urllib.quote(topic)
    req = tornado.httpclient.HTTPRequest(lookupd_url, method="GET", 
        connect_timeout=2, request_timeout=5)
    callback = functools.partial(_process_response, endpoint=endpoint, user_callback=user_callback)
    http_client().fetch(req, callback=callback)

def get_nsqd_stats(endpoint, user_callback):
    nsqd_url = endpoint + "/stats?format=json"
    req = tornado.httpclient.HTTPRequest(nsqd_url, method="GET", 
        connect_timeout=2, request_timeout=5)
    callback = functools.partial(_process_response, endpoint=endpoint, user_callback=user_callback)
    http_client().fetch(req, callback=callback)