#! /usr/bin/env python3 """ A basic RESTful server managing queues of messages. The server is started with the command: fifoserver.py interface-ip/port (e.g.: 127.0.0.1/12345) Warning: this script is only usable for test purposes with trusted clients since it does not address authentication aspects and does not prevent DoS attacks. Queues are kept in memory and are lost when the server is shutdown. Messages in the queues are never removed so the server memory footprint is damned to grow as the time goes... The exposed RESTful calls are the following (for each call we specify the list of supported fields sent using the urlencoding scheme in the body of the request); results are returned using a JSON dictionary and successful requests return the status code 200: - GET /: get a list of all the managed queues - POST /queueName: post a new message in the queue named queueName - Supported fields: - author: name of the author of the message (for this test server, the user is not authentified, so this field can easily be faked) - message: JSON-formatted message to post in the queue - Result of the request: {"id": id}; id is an integer uniquely designating the message in the queue (it is incremented for each new message) - Note: if no queue with the given name already exists, the queue is automatically created (and the message added get the id 0) - GET /queueName/id: get the message with the identifier id in the queue named queueName - Request fields: - timeout: timeout in seconds to retrieve the requested message - Result of the request: { "author": "authorOfTheMessage", "id": idOfTheMessage, "timestamp": UNIXTimestamp, "message": { ... } } - If the queue does not exist, this request immediately returns the HTTP status code 404 - If the queue exists but the message with the requested id have not already been put in the queue, we block until this message is available; however if the timeout occurs before the availability of the message, the status code 408 is returned Typically if we want to retrieve in real-time all the messages that were inserted in a given queue, one will emit a sequence of requests GET /queueName/0, GET /queueName/1, ..., GET /queueName/i... Versions log: - 2016/04/08: initial version @author chilowi at u-pem.fr """ import cgi import json import sys from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from threading import Lock, Condition from time import monotonic, time TIME_QUANTUM = 0.1 class HttpException(Exception): def __init__(self, code, message): self.code = code self.message = message class Queue(object): def __init__(self): self.content = [] self.lock = Lock() self.conditions = {} def __len__(self): return len(self.content) def append(self, data): with self.lock: self.content.append(data) cond = self.conditions.get(len(self.content) - 1) if cond: cond[0].notify_all() return len(self.content) - 1 def get(self, _id, timeout=None): start = monotonic() with self.lock: if _id < 0: return None elif _id < len(self.content): return self.content[_id] elif timeout == 0.0: return None else: # print("Timeout: {}".format(timeout)) (cond, waiters) = self.conditions.get(_id, (None, 0)) if cond is None: cond = Condition(self.lock) self.conditions[_id] = (cond, waiters + 1) try: while not cond.wait_for(lambda: _id < len(self.content), timeout=TIME_QUANTUM) \ and (not timeout or monotonic() - start < timeout): pass return self.content[_id] if _id < len(self.content) \ else None finally: (cond, waiters) = self.conditions[_id] if waiters == 1: # remove the now useless condition del self.conditions[_id] else: self.conditions[_id] = (cond, waiters - 1) class FifoHandlerFactory(object): def __init__(self): self.lock = Lock() self.queues = {} def post_message(self, queue_name, author, message): if not queue_name or not author or not message: raise HttpException(417, "All the fields were not supplied") try: message = json.loads(message) except: raise HttpException( 417, "The message field does not follow a JSON-format: {}".format( message)) with self.lock: q = self.queues.setdefault(queue_name, Queue()) _id = q.append( {"author": author, "timestamp": int(time()), "message": message}) return {"id": _id} def get_message(self, queue_name, _id, timeout): with self.lock: if queue_name not in self.queues: raise HttpException(404, "The queue {} is inexistent".format( queue_name)) q = self.queues[queue_name] if _id < 0: raise HttpException(404, "No message with negative id exists") r = q.get(_id, timeout) if r is None: raise HttpException(408, "Timeout occured when getting the message") else: return r def get_queues(self): """List all the queues""" with self.lock: return list(self.queues) def treat_request(self, request): from urllib.parse import urlparse url = urlparse(request.path) path = url.path path = path[1:].split("/") # Remove the starting slash queue_name = path[0] env = {"REQUEST_METHOD": request.command, "QUERY_STRING": url.query, "CONTENT_LENGTH": request.headers.get('Content-Length', -1), "CONTENT_TYPE": request.headers.get('Content-Type', None)} parsed = cgi.parse(request.rfile, env) result = None def get_field(name, integer=False): r = parsed.get(name) if not r: return None else: return r[0] if not integer else int(r[0]) try: if request.command == "POST": # post a message in a queue result = self.post_message(queue_name, get_field("author"), get_field("message")) elif request.command == "GET": if not queue_name: # list all the queues result = self.get_queues() else: try: _id = int(path[1]) except: raise HttpException( 417, "ID of the message must be an integer") # get a message in a queue result = self.get_message(queue_name, _id, int( get_field("timeout", integer=True))) except HttpException as e: request.send_response(e.code, e.message) request.end_headers() except Exception as e: request.send_response( 500, "An exception was encountered with the message {}".format(e)) request.end_headers() else: import json r2 = json.dumps(result).encode("UTF-8") request.send_response(200, 'OK') request.send_header('Content-Type', 'application/json') request.send_header('Content-Length', str(len(r2))) request.end_headers() request.wfile.write(r2) def get_handler(self): p = self class Handler(BaseHTTPRequestHandler): def do_GET(self): return p.treat_request(self) def do_POST(self): return p.treat_request(self) return Handler class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): pass def serve(address): h = ThreadingHTTPServer(address, FifoHandlerFactory().get_handler()) return h.serve_forever() if __name__ == "__main__": if len(sys.argv) < 2: print(__doc__) sys.exit(-1) else: (iface, port) = sys.argv[1].split('/', 1) serve((iface, int(port)))