import time, json, os, re import logging, threading from socket import socket, AF_INET, SOCK_STREAM, SHUT_WR, error as SocketError import socket from struct import pack, unpack from time import sleep from tornado import gen from tornado.ioloop import IOLoop from tornado.web import RequestHandler, Application, url, StaticFileHandler from rwlock import RWLock from ConfigParser import SafeConfigParser class ActionError(Exception): pass def add_monitor(mesg): if len(reclaimed_ids): jid = reclaimed_ids.pop() else: global id_cnt jid = id_cnt id_cnt += 1 if mesg.has_key("name"): name = mesg["name"] else: name = "monitor-{0}".format(jid) metadata = None if mesg.has_key("metadata"): metadata = mesg["metadata"] stat_res[jid] = {"name": name, "records": [], "rcnt": 0, "jid": jid, "metadata": metadata} return json.dumps(jid) def del_monitor(mesg): mid = check_id(mesg) del stat_res[mid] reclaimed_ids.append(mid) return "" def check_id(mesg): if mesg.has_key("jid"): try: jid = int(mesg["jid"]) except ValueError: raise ActionError("invalid value of jid field") else: raise ActionError("jid field not specified") if not stat_res.has_key(jid): raise ActionError("the jid does not exist") logging.info("jid: {0}".format(jid)) return jid def add_record(mesg): mid = check_id(mesg) if mesg.has_key("record"): rec = stat_res[mid]["records"] while len(rec) >= MAX_RECORDS: rec.pop(0) rec.append({'rid': stat_res[mid]["rcnt"], 'rec': mesg["record"]}) stat_res[mid]["rcnt"] += 1 return "" def alter_records(mesg): mid = check_id(mesg) if mesg.has_key("records"): stat_res[mid]["records"] = mesg["records"] return "" def clear_records(mesg): mid = check_id(mesg) stat_res[mid]["records"] = [] return "" def command_server(): global c, cmd_socket, is_exiting cmd_socket = socket.socket(AF_INET, SOCK_STREAM) cmd_socket.bind((HOST, SOCKET_PORT)) cmd_socket.listen(5) while not is_exiting.isSet(): logging.debug("accepting") (conn, addr) = cmd_socket.accept() logging.debug("accepted") received = conn.recv(4) if not received: continue length, = unpack(" MAX_LENGTH: logging.warning("invalid header: {0}".format(length)) conn.close() # invalid header continue try:# mesg = json.loads(conn.recv(length)) if not isinstance(mesg, dict): logging.warning("not a javascript object") continue if not mesg.has_key("action"): raise ActionError("action not specified") logging.info("action: {0}".format(mesg["action"])) try: lock.acquire_write() conn.send(action_map[mesg["action"]](mesg)) finally: lock.release() except ValueError: logging.warning("malformed json string") except ActionError as e: logging.warning(e) except KeyError as e: logging.warning("action not found: {0}".format(mesg["action"])) except SocketError as e: logging.warning("socket error: {0}".format(e)) finally: conn.close() def cmd_shutdown(): global is_exiting, cmd_socket, cmd is_exiting.set(); cmd_socket.close() socket.socket(AF_INET, SOCK_STREAM).connect((HOST, SOCKET_PORT)) cmd.join() class AJAXHandler(RequestHandler): @gen.coroutine def get(self): def grab_lock(self, callback=None): try: lock.acquire_read() self.write(stat_res) finally: lock.release() callback() yield gen.Task(grab_lock, self) if __name__ == '__main__': config = SafeConfigParser() config.read('server_settings.cfg') logging.basicConfig(level=logging.DEBUG, format='[%(levelname)-7s] (%(threadName)-10s) %(message)s',) MAX_LENGTH = 65536 MAX_RECORDS = 32 HTTP_PORT = config.get('HTTP', 'port') or 2333 SOCKET_PORT = config.get('socket', 'port') or 2334 HOST = '' lock = RWLock() stat_res = {} is_exiting = threading.Event() reclaimed_ids = [] id_cnt = 0 action_map = {"create": add_monitor, "drop": del_monitor, "add": add_record, "clear": clear_records, "alter": alter_records} cmd = threading.Thread(target=command_server, name="local") cmd.setDaemon(True) cmd.start() try: app = Application([url(r"/ajax", AJAXHandler), url(r'/()', StaticFileHandler, {'path': "./static/index.html"}), url(r'/(.*)', StaticFileHandler, {'path': "./static/"})]) app.listen(HTTP_PORT) IOLoop.current().start() except KeyboardInterrupt: cmd_shutdown()