spothole/server/sse_broadcaster.py

import logging
import threading

from tornado.ioloop import IOLoop


class SSEBroadcaster:
    """Bridge between DataStore listener callbacks (which fire on provider threads) to Tornado's async SSE handlers
    (which live on the IOLoop thread) to avoid any interdependency between them."""

    def __init__(self):
        self._handlers = set()
        self._lock = threading.Lock()
        self._loop = None

    def bind_to_web_server_loop(self):
        self._loop = IOLoop.current()

    def register(self, handler):
        with self._lock:
            self._handlers.add(handler)

    def unregister(self, handler):
        with self._lock:
            self._handlers.discard(handler)

    def publish(self, value):
         self._loop.add_callback(self._broadcast, value)

    def _broadcast(self, value):
        with self._lock:
            handlers = list(self._handlers)
        for handler in handlers:
            try:
                handler.callback(value)
            except Exception:
                # Connection probably dropped, ignore and de-register the handler to stop getting future items.
                logging.debug("Failed to push to an SSE client; dropping it")
                self.unregister(handler)