#!/usr/bin/env python3 """Multi-room chat met Server-Sent Events. Stdlib only, single file, 3.6+. Run: python3 sse-chat.py --port 8082 Open http://127.0.0.1:8082/ in twee browsers en chat. Geen JS-deps: alleen EventSource + fetch. Spec (dit bestand is de spec): Endpoint Method Statusen Beschrijving /api/rooms GET 200 Kamers + aantallen /api/rooms POST 201/400 Kamer aanmaken {"room"} /api/rooms/{room}/messages GET 200/404 History (?since=&limit=) /api/rooms/{room}/messages POST 201/400/404 Bericht {"nick","text"} /events?room={room} GET 200/404 SSE-stream (id/event/retry) / GET 200 Embedded UI /healthz GET 200 {"ok":true,...} SSE-events: "message" (JSON-bericht), "hello" (laatste id bij connect). Reconnect: Last-Event-ID-header of ?since= wordt gerespecteerd, gemiste berichten worden nagezonden. Heartbeat: ": ping" elke 15 s. Kamernamen: ^[a-z0-9-]{1,32}$. Standaardkamers: lobby, random, tech. Caps: MAX_ROOMS 50, MAX_HISTORY 200/room, MAX_NICK 32, MAX_TEXT 2000, MAX_BODY 8 KiB, HEARTBEAT 15 s. """ import argparse import json import re import socketserver import threading import time from collections import deque from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import parse_qs, urlparse try: from queue import Queue, Empty except ImportError: # Python 2 fallback, nooit gebruikt maar goedkoop from Queue import Queue, Empty ROOM_RE = re.compile(r"^[a-z0-9-]{1,32}$") class ChatError(Exception): def __init__(self, status, message): super(ChatError, self).__init__(message) self.status = status self.message = message class ChatStore(object): """Kamers, history en SSE-abonnees. Alles achter een RLock.""" MAX_ROOMS = 50 MAX_HISTORY = 200 MAX_SUBSCRIBERS = 500 DEFAULT_ROOMS = ("lobby", "random", "tech") def __init__(self): self._lock = threading.RLock() self._rooms = {} self._subs = {} for room in self.DEFAULT_ROOMS: self._rooms[room] = {"messages": deque(maxlen=self.MAX_HISTORY), "next_id": 1} @staticmethod def _now(): return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def rooms(self): with self._lock: return [{"room": name, "messages": len(info["messages"])} for name, info in sorted(self._rooms.items())] def create_room(self, room): if not isinstance(room, str) or not ROOM_RE.match(room): raise ChatError(400, "bad room (want [a-z0-9-]{1,32})") with self._lock: if room in self._rooms: raise ChatError(409, "room exists") if len(self._rooms) >= self.MAX_ROOMS: raise ChatError(507, "too many rooms") self._rooms[room] = {"messages": deque(maxlen=self.MAX_HISTORY), "next_id": 1} return {"room": room, "messages": 0} def _require(self, room): if room not in self._rooms: raise ChatError(404, "no such room") return self._rooms[room] def history(self, room, since=0, limit=100): with self._lock: info = self._require(room) items = [m for m in info["messages"] if m["id"] > since] return items[-limit:] def post(self, room, nick, text): if not isinstance(nick, str) or not 1 <= len(nick.strip()) <= 32: raise ChatError(400, "nick must be 1..32 chars") if not isinstance(text, str) or not 1 <= len(text.strip()) <= 2000: raise ChatError(400, "text must be 1..2000 chars") with self._lock: info = self._require(room) msg = {"id": info["next_id"], "room": room, "nick": nick.strip()[:32], "text": text.strip()[:2000], "ts": self._now()} info["next_id"] += 1 info["messages"].append(msg) subs = list(self._subs.get(room, [])) for queue in subs: try: queue.put_nowait(msg) except Exception: pass return msg def subscribe(self, room): with self._lock: self._require(room) total = sum(len(v) for v in self._subs.values()) if total >= self.MAX_SUBSCRIBERS: raise ChatError(503, "too many subscribers") queue = Queue(maxsize=200) self._subs.setdefault(room, []).append(queue) return queue def unsubscribe(self, room, queue): with self._lock: subs = self._subs.get(room, []) if queue in subs: subs.remove(queue) class ChatHandler(BaseHTTPRequestHandler): server_version = "TaiChat/1.0" protocol_version = "HTTP/1.1" MAX_BODY = 8 * 1024 HEARTBEAT = 15 def log_message(self, fmt, *args): pass def _send(self, code, body=b"", ctype="text/plain; charset=utf-8"): if isinstance(body, str): body = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() if self.command != "HEAD" and body: self.wfile.write(body) def _json(self, code, obj): self._send(code, json.dumps(obj), "application/json") def _read_json(self): try: length = int(self.headers.get("Content-Length", "0") or "0") except ValueError: raise ChatError(400, "bad Content-Length") if length < 0 or length > self.MAX_BODY: raise ChatError(413, "body too large") raw = self.rfile.read(length) if length else b"" try: data = json.loads((raw or b"{}").decode("utf-8")) except (UnicodeDecodeError, ValueError): raise ChatError(400, "body is not valid JSON") if not isinstance(data, dict): raise ChatError(400, "body must be a JSON object") return data def _events(self, room, since): store = self.server.chat # type: ChatStore try: queue = store.subscribe(room) except ChatError as exc: self._json(exc.status, {"error": exc.message}) return self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.send_header("X-Accel-Buffering", "no") self.end_headers() try: self._sse("hello", {"room": room, "since": since}) for msg in store.history(room, since=since, limit=200): self._sse("message", msg, msg["id"]) while True: try: msg = queue.get(timeout=self.HEARTBEAT) except Empty: self.wfile.write(b": ping\n\n") self.wfile.flush() continue if msg["id"] <= since: continue since = msg["id"] self._sse("message", msg, msg["id"]) except (BrokenPipeError, ConnectionResetError, OSError): pass finally: store.unsubscribe(room, queue) def _sse(self, event, obj, msg_id=None): lines = ["event: %s" % event] if msg_id is not None: lines.append("id: %d" % msg_id) lines.append("retry: 3000") for chunk in json.dumps(obj).split("\n"): lines.append("data: %s" % chunk) self.wfile.write(("\n".join(lines) + "\n\n").encode("utf-8")) self.wfile.flush() def _handle(self): store = self.server.chat # type: ChatStore parsed = urlparse(self.path) parts = [p for p in parsed.path.split("/") if p] method = self.command if not parts: if method != "GET": raise ChatError(405, "method not allowed") self._send(200, CHAT_HTML, "text/html; charset=utf-8") return if parts == ["healthz"]: rooms = store.rooms() self._json(200, {"ok": True, "rooms": len(rooms), "messages": sum(r["messages"] for r in rooms)}) return if parts == ["events"] and method == "GET": qs = parse_qs(parsed.query) room = qs.get("room", ["lobby"])[0] since = qs.get("since", ["0"])[0] last_id = self.headers.get("Last-Event-ID", "").strip() if last_id.isdigit(): since = last_id try: since = max(0, int(since)) except ValueError: raise ChatError(400, "bad since") self._events(room, since) return if parts[0] != "api" or len(parts) < 2 or parts[1] != "rooms": raise ChatError(404, "unknown endpoint") if len(parts) == 2: if method == "GET": self._json(200, {"rooms": store.rooms()}) elif method == "POST": data = self._read_json() self._json(201, store.create_room(data.get("room", ""))) else: raise ChatError(405, "method not allowed") return room = parts[2] if len(parts) == 4 and parts[3] == "messages": if method == "GET": qs = parse_qs(parsed.query) try: since = max(0, int(qs.get("since", ["0"])[0])) limit = min(max(int(qs.get("limit", ["100"])[0]), 1), 200) except ValueError: raise ChatError(400, "bad since/limit") self._json(200, {"room": room, "messages": store.history( room, since=since, limit=limit)}) elif method == "POST": data = self._read_json() self._json(201, store.post(room, data.get("nick", ""), data.get("text", ""))) else: raise ChatError(405, "method not allowed") return raise ChatError(404, "unknown endpoint") def _dispatch(self): try: self._handle() except ChatError as exc: self._json(exc.status, {"error": exc.message}) except BrokenPipeError: pass do_GET = _dispatch do_HEAD = _dispatch do_POST = _dispatch CHAT_HTML = """ Chat

#lobby

verbinden...
""" class ChatServer(socketserver.ThreadingMixIn, HTTPServer): daemon_threads = True allow_reuse_address = True def main(argv=None): parser = argparse.ArgumentParser( description="Multi-room SSE chat (stdlib only).") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8082) args = parser.parse_args(argv) server = ChatServer((args.host, args.port), ChatHandler) server.chat = ChatStore() print("chat: http://%s:%d/" % (args.host, args.port)) try: server.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()