spothole/server/webserver.py

import asyncio
import logging
import os

import tornado
from tornado.web import StaticFileHandler

from core.utils import empty_queue
from server.handlers.api.addspot import APISpotHandler
from server.handlers.api.alerts import APIAlertsHandler, APIAlertsStreamHandler
from server.handlers.api.lookups import APILookupCallHandler, APILookupSIGRefHandler, APILookupGridHandler
from server.handlers.api.options import APIOptionsHandler
from server.handlers.api.spots import APISpotsHandler, APISpotsStreamHandler
from server.handlers.api.status import APIStatusHandler
from server.handlers.metrics import PrometheusMetricsHandler
from server.handlers.pagetemplate import PageTemplateHandler


class WebServer:
    """Provides the public-facing web server."""

    def __init__(self, spots, alerts, status_data, port):
        """Constructor"""

        self._spots = spots
        self._alerts = alerts
        self._sse_spot_queues = []
        self._sse_alert_queues = []
        self._status_data = status_data
        self._port = port
        self._shutdown_event = asyncio.Event()
        self.web_server_metrics = {
            "last_page_access_time": None,
            "last_api_access_time": None,
            "page_access_counter": 0,
            "api_access_counter": 0,
            "status": "Starting"
        }

    def start(self):
        """Start the web server"""

        asyncio.run(self._start_inner())

    def stop(self):
        """Stop the web server"""

        self._shutdown_event.set()

    async def _start_inner(self):
        """Start method (async). Sets up the Tornado application."""

        app = tornado.web.Application([
            # Routes for API calls
            (r"/api/v1/spots", APISpotsHandler, {"spots": self._spots, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/alerts", APIAlertsHandler,
             {"alerts": self._alerts, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/spots/stream", APISpotsStreamHandler,
             {"sse_spot_queues": self._sse_spot_queues, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/alerts/stream", APIAlertsStreamHandler,
             {"sse_alert_queues": self._sse_alert_queues, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/options", APIOptionsHandler,
             {"status_data": self._status_data, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/status", APIStatusHandler,
             {"status_data": self._status_data, "web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/lookup/call", APILookupCallHandler, {"web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/lookup/sigref", APILookupSIGRefHandler, {"web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/lookup/grid", APILookupGridHandler, {"web_server_metrics": self.web_server_metrics}),
            (r"/api/v1/spot", APISpotHandler, {"spots": self._spots, "web_server_metrics": self.web_server_metrics}),
            # Routes for templated pages
            (r"/", PageTemplateHandler, {"template_name": "spots", "web_server_metrics": self.web_server_metrics}),
            (r"/map", PageTemplateHandler, {"template_name": "map", "web_server_metrics": self.web_server_metrics}),
            (r"/bands", PageTemplateHandler, {"template_name": "bands", "web_server_metrics": self.web_server_metrics}),
            (r"/alerts", PageTemplateHandler,
             {"template_name": "alerts", "web_server_metrics": self.web_server_metrics}),
            (r"/add-spot", PageTemplateHandler,
             {"template_name": "add_spot", "web_server_metrics": self.web_server_metrics}),
            (r"/status", PageTemplateHandler,
             {"template_name": "status", "web_server_metrics": self.web_server_metrics}),
            (r"/about", PageTemplateHandler, {"template_name": "about", "web_server_metrics": self.web_server_metrics}),
            (r"/apidocs", PageTemplateHandler,
             {"template_name": "apidocs", "web_server_metrics": self.web_server_metrics}),
            # Route for Prometheus metrics
            (r"/metrics", PrometheusMetricsHandler),
            # Default route to serve from "webassets"
            (r"/(.*)", StaticFileHandler, {"path": os.path.join(os.path.dirname(__file__), "../webassets")}),
        ],
            template_path=os.path.join(os.path.dirname(__file__), "../templates"),
            debug=False)
        app.listen(self._port)
        await self._shutdown_event.wait()

    def notify_new_spot(self, spot):
        """Internal method called when a new spot is added to the system. This is used to ping any SSE clients that are
        awaiting a server-sent message with new spots."""

        for queue in self._sse_spot_queues:
            try:
                queue.put(spot)
            except:
                # Cleanup thread was probably deleting the queue, that's fine
                pass
        pass

    def notify_new_alert(self, alert):
        """Internal method called when a new alert is added to the system. This is used to ping any SSE clients that are
        awaiting a server-sent message with new spots."""

        for queue in self._sse_alert_queues:
            try:
                queue.put(alert)
            except:
                # Cleanup thread was probably deleting the queue, that's fine
                pass
        pass

    def clean_up_sse_queues(self):
        """Clean up any SSE queues that are growing too large; probably their client disconnected and we didn't catch it
        properly for some reason."""

        for q in self._sse_spot_queues:
            try:
                if q.full():
                    logging.warning(
                        "A full SSE spot queue was found, presumably because the client disconnected strangely. It has been removed.")
                    self._sse_spot_queues.remove(q)
                    empty_queue(q)
            except:
                # Probably got deleted already on another thread
                pass
        for q in self._sse_alert_queues:
            try:
                if q.full():
                    logging.warning(
                        "A full SSE alert queue was found, presumably because the client disconnected strangely. It has been removed.")
                    self._sse_alert_queues.remove(q)
                    empty_queue(q)
            except:
                # Probably got deleted already on another thread
                pass
        pass