Refactor of caching & data storage part 2 #118

alertproviders/alert_provider.py

@@ -2,7 +2,7 @@ from datetime import datetime
 
 import pytz
 
-from core.config import MAX_ALERT_AGE
+from core.data_store import DATA_STORE
 
 
 class AlertProvider:
@@ -15,14 +15,7 @@ class AlertProvider:
         self.enabled = provider_config["enabled"]
         self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
         self.status = "Not Started" if self.enabled else "Disabled"
-        self._alerts = None
+        self._alerts = DATA_STORE.alerts
-        self._web_server = None
-
-    def setup(self, alerts, web_server):
-        """Set up the provider, e.g. giving it the alert list to work from"""
-
-        self._alerts = alerts
-        self._web_server = web_server
 
     def start(self):
         """Start the provider. This should return immediately after spawning threads to access the remote resources"""
@@ -45,9 +38,6 @@ class AlertProvider:
     def _add_alert(self, alert):
         if not alert.expired():
             self._alerts.set(alert.id, alert)
-            # Ping the web server in case we have any SSE connections that need to see this immediately
-            if self._web_server:
-                self._web_server.notify_new_alert(alert)
 
     def stop(self):
         """Stop any threads and prepare for application shutdown"""

core/live_data_cache.py

@@ -7,14 +7,17 @@ from cachetools import TTLCache
 
 
 class LiveDataCache:
-    """Cache for spots and alerts. Uses the faster in-memory TTLCache for normal data I/O, including the TTL to enforce
+    """Cache for spots and alerts. Uses the fast in-memory TTLCache for normal data I/O, including the TTL to enforce
     maximum lifetime, and adds a separate diskcache to which we can save and load the TTLCache to provide persistence.
-    Also adds thread safety which TTLCache doesn't do."""
+    Also adds thread safety so spots and alerts can come from any thread, and a listener mechanism so the web server
+    can get a callback when new spots/alerts are added, and send them to any SSE clients."""
 
     def __init__(self, maxsize, ttl, snapshot_dir, snapshot_interval_sec):
         self._cache = TTLCache(maxsize=maxsize, ttl=ttl)
         self._lock = threading.Lock()
         self._ttl = ttl
+        self._listeners = []
+        self._listeners_lock = threading.Lock()
         self._snapshot_dir = snapshot_dir
         self._disk_cache = diskcache.Cache(str(snapshot_dir))
         self._load_snapshot()
@@ -24,6 +27,16 @@ class LiveDataCache:
         with self._lock:
             self._cache[key] = value
 
+        # Notify listeners
+        with self._listeners_lock:
+            listeners = list(self._listeners)
+        for callback in listeners:
+            try:
+                callback(value)
+            except Exception:
+                logging.error("Listener raised an exception for key %s", key, exc_info=True)
+
+
     def get(self, key, default=None):
         with self._lock:
             return self._cache.get(key, default)
@@ -32,10 +45,25 @@ class LiveDataCache:
         with self._lock:
             self._cache.pop(key, None)
 
+    def keys(self):
+        with self._lock:
+            return list(self._cache.keys())
+
     def values(self):
         with self._lock:
             return list(self._cache.values())
 
+    def add_listener(self, callback):
+        """Register callback(value) which will be called whenever a new spot/alert item is added via set(). Used by the
+        web server (via SSEBroadcaster) to send SSE clients an update on every new spot."""
+
+        with self._listeners_lock:
+            self._listeners.append(callback)
+
+    def remove_listener(self, callback):
+        with self._listeners_lock:
+            self._listeners.remove(callback)
+
     def save_snapshot(self):
         with self._lock:
             # Store the time with the data so we can avoid loading anything nxt time that's older than TTL

core/lookup_helper.py

@@ -14,10 +14,10 @@ from pyhamtools.locator import latlong_to_locator
 from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout
 from requests_cache import CachedSession
 
-from core.url_data_cache import URL_DATA_CACHE
 from core.config import config
 from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, \
     HTTP_HEADERS, HAMQTH_PRG, MODE_ALIASES
+from core.url_data_cache import URL_DATA_CACHE
 
 # QRZ XML field names differ from pyhamtools' normalised names; map them here.
 _QRZ_FIELD_MAP = {

core/status_reporter.py

@@ -7,16 +7,16 @@ import pytz
 
 from core.config import SERVER_OWNER_CALLSIGN
 from core.constants import SOFTWARE_VERSION
+from core.data_store import DATA_STORE
 from core.prometheus_metrics_handler import memory_use_gauge, spots_gauge, alerts_gauge
 
 
 class StatusReporter:
     """Provides a timed update of the application's status data."""
 
-    def __init__(self, data_store, run_interval, web_server,spot_providers, alert_providers, solar_condition_providers):
+    def __init__(self, run_interval, web_server,spot_providers, alert_providers, solar_condition_providers):
         """Constructor"""
 
-        self._data_store = data_store
         self._run_interval = run_interval
         self._web_server = web_server
         self._spot_providers = spot_providers
@@ -26,8 +26,8 @@ class StatusReporter:
         self._stop_event = Event()
         self._startup_time = datetime.now(pytz.UTC)
 
-        self._data_store.status_data["software-version"] = SOFTWARE_VERSION
+        DATA_STORE.status_data["software-version"] = SOFTWARE_VERSION
-        self._data_store.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN
+        DATA_STORE.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN
 
     def start(self):
         """Start the reporter thread"""
@@ -51,28 +51,28 @@ class StatusReporter:
     def _report(self):
         """Write status information"""
 
-        self._data_store.status_data["uptime"] = (datetime.now(pytz.UTC) - self._startup_time).total_seconds()
+        DATA_STORE.status_data["uptime"] = (datetime.now(pytz.UTC) - self._startup_time).total_seconds()
-        self._data_store.status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3)
+        DATA_STORE.status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3)
-        self._data_store.status_data["num_spots"] = len(self._data_store.spots.values())
+        DATA_STORE.status_data["num_spots"] = len(DATA_STORE.spots.values())
-        self._data_store.status_data["num_alerts"] = len(self._data_store.alerts.values())
+        DATA_STORE.status_data["num_alerts"] = len(DATA_STORE.alerts.values())
-        self._data_store.status_data["spot_providers"] = list(
+        DATA_STORE.status_data["spot_providers"] = list(
         map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status,
                        "last_updated": p.last_update_time.replace(
                            tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0,
                        "last_spot": p.last_spot_time.replace(
                            tzinfo=pytz.UTC).timestamp() if p.last_spot_time.year > 2000 else 0},
             self._spot_providers))
-        self._data_store.status_data["alert_providers"] = list(
+        DATA_STORE.status_data["alert_providers"] = list(
         map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status,
                        "last_updated": p.last_update_time.replace(
                            tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0},
             self._alert_providers))
-        self._data_store.status_data["solar_condition_providers"] = list(
+        DATA_STORE.status_data["solar_condition_providers"] = list(
         map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status,
                        "last_updated": p.last_update_time.replace(
                            tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0},
             self._solar_condition_providers))
-        self._data_store.status_data["webserver"] = {"status": self._web_server.web_server_metrics["status"],
+        DATA_STORE.status_data["webserver"] = {"status": self._web_server.web_server_metrics["status"],
                                       "last_api_access": self._web_server.web_server_metrics[
                                           "last_api_access_time"].replace(
                                           tzinfo=pytz.UTC).timestamp() if self._web_server.web_server_metrics[
@@ -87,5 +87,5 @@ class StatusReporter:
 
         # Update Prometheus metrics
         memory_use_gauge.set(psutil.Process(os.getpid()).memory_info().rss)
-        spots_gauge.set(len(self._data_store.spots.values()))
+        spots_gauge.set(len(DATA_STORE.spots.values()))
-        alerts_gauge.set(len(self._data_store.alerts.values()))
+        alerts_gauge.set(len(DATA_STORE.alerts.values()))

core/utils.py

@@ -6,13 +6,3 @@ def safe_json_dumps(obj):
     which are invalid in JSON."""
 
     return simplejson.dumps(obj, ensure_ascii=False, ignore_nan=True, default=lambda o: o.__dict__)
-
-
-def empty_queue(q):
-    """Empty a queue"""
-
-    while not q.empty():
-        try:
-            q.get_nowait()
-        except:
-            break

server/handlers/api/addspot.py

@@ -8,7 +8,7 @@ import tornado
 from tornado import httputil
 from tornado.web import Application
 
-from core.config import ALLOW_SPOTTING, MAX_SPOT_AGE
+from core.config import ALLOW_SPOTTING
 from core.constants import UNKNOWN_BAND
 from core.lookup_helper import infer_band_from_freq
 from core.prometheus_metrics_handler import api_requests_counter

server/handlers/api/alerts.py

@@ -1,7 +1,6 @@
 import copy
 import logging
 from datetime import datetime
-from queue import Queue
 from typing import Any
 
 import pytz
@@ -11,12 +10,9 @@ from tornado import httputil
 from tornado.web import Application
 
 from core.prometheus_metrics_handler import api_requests_counter
-from core.utils import safe_json_dumps, empty_queue
+from core.utils import safe_json_dumps
 from data.lookup_credentials import extract_credentials
 
-SSE_HANDLER_MAX_QUEUE_SIZE = 100
-SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000
-
 
 class APIAlertsHandler(tornado.web.RequestHandler):
     """API request handler for /api/v1/alerts"""
@@ -73,16 +69,14 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
     """API request handler for /api/v1/alerts/stream"""
 
     def __init__(self, application, request, **kwargs: Any):
-        self._sse_alert_queues = None
+        self._sse_alert_broadcaster = None
         self._web_server_metrics = None
         self._query_params = None
         self._credentials = None
-        self._alert_queue = None
-        self._heartbeat = None
         super().__init__(application, request, **kwargs)
 
-    def initialize(self, sse_alert_queues, web_server_metrics):
+    def initialize(self, _sse_alert_broadcaster, web_server_metrics):
-        self._sse_alert_queues = sse_alert_queues
+        self._sse_alert_broadcaster = _sse_alert_broadcaster
         self._web_server_metrics = web_server_metrics
 
     def custom_headers(self):
@@ -104,59 +98,32 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
             self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
             self._credentials = extract_credentials(self._query_params)
 
-            # Create a alert queue and add it to the web server's list. The web server will fill this when alerts arrive
-            self._alert_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE)
-            self._sse_alert_queues.append(self._alert_queue)
-
-            # Set up a timed callback to check if anything is in the queue
-            self._heartbeat = tornado.ioloop.PeriodicCallback(self._callback, SSE_HANDLER_QUEUE_CHECK_INTERVAL)
-            self._heartbeat.start()
-
             # Flush headers immediately so nginx doesn't time out waiting for a response
             self.write_message("keepalive", "")
 
+            # Register to handle new alerts arriving. The callback() method will get called with the new alert as an
+            # argument.
+            self._sse_alert_broadcaster.register(self)
+
         except Exception as e:
             logging.warning("Exception when serving SSE socket: %s", e, exc_info=True)
             self.close()
 
     def close(self):
-        """When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it"""
+        """When the user closes the socket, deregister ourselves from the alert broadcaster"""
 
-        try:
+        self._sse_alert_broadcaster.unregister(self)
-            if self._alert_queue in self._sse_alert_queues:
-                self._sse_alert_queues.remove(self._alert_queue)
-            empty_queue(self._alert_queue)
-        except:
-            pass
-        try:
-            self._heartbeat.stop()
-        except:
-            pass
-        self._alert_queue = None
         super().close()
 
-    def _callback(self):
+    def callback(self, alert):
-        """Callback to check if anything has arrived in the queue, and if so send it to the client"""
+        """Callback when a new alert arrives"""
 
         try:
-            if self._alert_queue:
-                if not self._alert_queue.empty():
-                    while not self._alert_queue.empty():
-                        alert = self._alert_queue.get()
-                        # If the new alert matches our param filters, send it to the client. If not, ignore it.
             if alert_allowed_by_query(alert, self._query_params):
                 if self._credentials:
                     alert = copy.deepcopy(alert)
                     alert.infer_missing(self._credentials)
                 self.write_message(msg=safe_json_dumps(alert))
-
-                else:
-                    # Send a keepalive comment if the queue was empty
-                    self.write_message("keepalive", "")
-
-                if self._alert_queue not in self._sse_alert_queues:
-                    logging.error("Web server cleared up a queue of an active connection!")
-                    self.close()
         except Exception as e:
             logging.warning("Exception in SSE callback, connection will be closed: %s", e, exc_info=True)
             self.close()
@@ -169,7 +136,7 @@ def get_alert_list_with_filters(all_alerts, query):
     # Create a shallow copy of the alert list ordered by start time, then filter the list to reduce it only to alerts
     # that match the filter parameters in the query string. Finally, apply a limit to the number of alerts returned.
     # The list of query string filters is defined in the API docs.
-    alert_ids = list(all_alerts.iterkeys())
+    alert_ids = all_alerts.keys()
     alerts = []
     for k in alert_ids:
         a = all_alerts.get(k)

server/handlers/api/dxstats.py

@@ -40,7 +40,7 @@ class APIDxStatsHandler(tornado.web.RequestHandler):
             one_hour_ago = (datetime.now(pytz.UTC) - timedelta(hours=1)).timestamp()
             counts = Counter()
 
-            for key in self._spots.iterkeys():
+            for key in self._spots.keys():
                 spot = self._spots.get(key)
                 if spot is None:
                     continue

server/handlers/api/spots.py

@@ -1,7 +1,6 @@
 import copy
 import logging
 from datetime import datetime, timedelta
-from queue import Queue
 from typing import Any
 
 import pytz
@@ -11,12 +10,9 @@ from tornado import httputil
 from tornado.web import Application
 
 from core.prometheus_metrics_handler import api_requests_counter
-from core.utils import safe_json_dumps, empty_queue
+from core.utils import safe_json_dumps
 from data.lookup_credentials import extract_credentials
 
-SSE_HANDLER_MAX_QUEUE_SIZE = 1000
-SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000
-
 
 class APISpotsHandler(tornado.web.RequestHandler):
     """API request handler for /api/v1/spots"""
@@ -73,16 +69,14 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
     """API request handler for /api/v1/spots/stream"""
 
     def __init__(self, application, request, **kwargs: Any):
-        self._sse_spot_queues = None
+        self._sse_spot_broadcaster = None
         self._web_server_metrics = None
         self._query_params = None
         self._credentials = None
-        self._spot_queue = None
-        self._heartbeat = None
         super().__init__(application, request, **kwargs)
 
-    def initialize(self, sse_spot_queues, web_server_metrics):
+    def initialize(self, sse_spot_broadcaster, web_server_metrics):
-        self._sse_spot_queues = sse_spot_queues
+        self._sse_spot_broadcaster = sse_spot_broadcaster
         self._web_server_metrics = web_server_metrics
 
     def custom_headers(self):
@@ -106,59 +100,33 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
             self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
             self._credentials = extract_credentials(self._query_params)
 
-            # Create a spot queue and add it to the web server's list. The web server will fill this when spots arrive
-            self._spot_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE)
-            self._sse_spot_queues.append(self._spot_queue)
-
-            # Set up a timed callback to check if anything is in the queue
-            self._heartbeat = tornado.ioloop.PeriodicCallback(self._callback, SSE_HANDLER_QUEUE_CHECK_INTERVAL)
-            self._heartbeat.start()
-
             # Flush headers immediately so nginx doesn't time out waiting for a response
             self.write_message("keepalive", "")
 
+            # Register to handle new spots arriving. The callback() method will get called with the new spot as an
+            # argument.
+            self._sse_spot_broadcaster.register(self)
+
         except Exception as e:
             logging.warning("Exception when serving SSE socket: %s", e, exc_info=True)
             self.close()
 
     def close(self):
-        """When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it"""
+        """When the user closes the socket, deregister ourselves from the spot broadcaster"""
 
-        try:
+        self._sse_spot_broadcaster.unregister(self)
-            if self._spot_queue in self._sse_spot_queues:
-                self._sse_spot_queues.remove(self._spot_queue)
-            empty_queue(self._spot_queue)
-        except:
-            pass
-        try:
-            self._heartbeat.stop()
-        except:
-            pass
-        self._spot_queue = None
         super().close()
 
-    def _callback(self):
+    def callback(self, spot):
-        """Callback to check if anything has arrived in the queue, and if so send it to the client"""
+        """Callback when a new spot arrives"""
 
         try:
-            if self._spot_queue:
-                if not self._spot_queue.empty():
-                    while not self._spot_queue.empty():
-                        spot = self._spot_queue.get()
             # If the new spot matches our param filters, send it to the client. If not, ignore it.
             if spot_allowed_by_query(spot, self._query_params):
                 if self._credentials:
                     spot = copy.deepcopy(spot)
                     spot.infer_missing(self._credentials)
                 self.write_message(msg=safe_json_dumps(spot))
-
-                else:
-                    # Send a keepalive comment if the queue was empty
-                    self.write_message("keepalive", "")
-
-                if self._spot_queue not in self._sse_spot_queues:
-                    logging.error("Web server cleared up a queue of an active connection!")
-                    self.close()
         except Exception as e:
             logging.warning("Exception in SSE callback, connection will be closed: %s", e, exc_info=True)
             self.close()
@@ -171,7 +139,7 @@ def get_spot_list_with_filters(all_spots, query):
     # Create a shallow copy of the spot list, ordered by spot time, then filter the list to reduce it only to spots
     # that match the filter parameters in the query string. Finally, apply a limit to the number of spots returned.
     # The list of query string filters is defined in the API docs.
-    spot_ids = list(all_spots.iterkeys())
+    spot_ids = all_spots.keys()
     spots = []
     for k in spot_ids:
         s = all_spots.get(k)

server/sse_broadcaster.py

@@ -0,0 +1,36 @@
+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 = 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._fan_out, value)
+
+    def _fan_out(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)

server/webserver.py

@@ -6,7 +6,7 @@ import tornado
 from tornado.web import StaticFileHandler
 
 from core.config import ALLOW_SPOTTING, WEB_SERVER_PORT, API_ONLY_MODE, LOG_WEB_REQUESTS, BASE_URL
-from core.utils import empty_queue
+from core.data_store import DATA_STORE
 from server.handlers.api.addspot import APISpotHandler
 from server.handlers.api.alerts import APIAlertsHandler, APIAlertsStreamHandler
 from server.handlers.api.dxstats import APIDxStatsHandler
@@ -18,6 +18,7 @@ from server.handlers.api.status import APIStatusHandler
 from server.handlers.manifesthandler import ManifestHandler
 from server.handlers.metrics import PrometheusMetricsHandler
 from server.handlers.pagetemplate import PageTemplateHandler
+from server.sse_broadcaster import SSEBroadcaster
 
 _HERE = os.path.dirname(__file__ or "")
 
@@ -25,12 +26,12 @@ _HERE = os.path.dirname(__file__ or "")
 class WebServer:
     """Provides the public-facing web server."""
 
-    def __init__(self, data_store):
+    def __init__(self):
         """Constructor"""
 
-        self._data_store = data_store
+        self._data_store = DATA_STORE
-        self._sse_spot_queues = []
+        self._spot_broadcaster = SSEBroadcaster()
-        self._sse_alert_queues = []
+        self._alert_broadcaster = SSEBroadcaster()
         self._port = WEB_SERVER_PORT
         self._api_only_mode = API_ONLY_MODE
         self._shutdown_event = asyncio.Event()
@@ -42,6 +43,10 @@ class WebServer:
             "status": "Starting"
         }
 
+        # Listen for new spots and alerts being added to the cache, so we can notify SSE clients immediately
+        DATA_STORE.spots.add_listener(self._spot_broadcaster.publish)
+        DATA_STORE.alerts.add_listener(self._alert_broadcaster.publish)
+
     def start(self):
         """Start the web server"""
 
@@ -64,13 +69,13 @@ class WebServer:
             (r"/api/v1/spots", APISpotsHandler, {"spots": self._data_store.spots, **handler_opts}),
             (r"/api/v1/alerts", APIAlertsHandler, {"alerts": self._data_store.alerts, **handler_opts}),
             (r"/api/v1/spots/stream", APISpotsStreamHandler,
-             {"sse_spot_queues": self._sse_spot_queues, **handler_opts}),
+             {"sse_spot_broadcaster": self._spot_broadcaster, **handler_opts}),
             (r"/api/v1/alerts/stream", APIAlertsStreamHandler,
-             {"sse_alert_queues": self._sse_alert_queues, **handler_opts}),
+             {"sse_alert_broadcaster": self._alert_broadcaster, **handler_opts}),
             (r"/api/v1/solar", APISolarConditionsHandler, {"solar_conditions": self._data_store.solar, **handler_opts}),
             (r"/api/v1/dxstats", APIDxStatsHandler, {"spots": self._data_store.spots, **handler_opts}),
-            (r"/api/v1/options", APIOptionsHandler, {"status_data": self._data_store.status, **handler_opts}),
+            (r"/api/v1/options", APIOptionsHandler, {"status_data": self._data_store.status_data, **handler_opts}),
-            (r"/api/v1/status", APIStatusHandler, {"status_data": self._data_store.status, **handler_opts}),
+            (r"/api/v1/status", APIStatusHandler, {"status_data": self._data_store.status_data, **handler_opts}),
             (r"/api/v1/lookup/call", APILookupCallHandler, {**handler_opts}),
             (r"/api/v1/lookup/sigref", APILookupSIGRefHandler, {**handler_opts}),
             (r"/api/v1/lookup/grid", APILookupGridHandler, {**handler_opts}),
@@ -118,55 +123,6 @@ class WebServer:
         logging.info("You can access your copy of Spothole at " + BASE_URL)
         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
 
 def request_log(handler):
     """Custom log function to provide more data about requests when enabled, and to provide the ability to turn off

sigrefdataproviders/sig_ref_data_provider.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+
+import pytz
+
+
+class SIGRefDataProvider:
+    """Generic SIG reference data provider class. Subclasses of this query the individual URLs or files for data."""
+
+    def __init__(self, name, provider_config):
+        """Constructor"""
+
+        self.name = name
+        self.enabled = provider_config["enabled"]
+        self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
+        self.last_spot_time = datetime.min.replace(tzinfo=pytz.UTC)
+        self.status = "Not Started" if self.enabled else "Disabled"
+
+
+    def start(self):
+        """Start the provider. This should return immediately after spawning threads to access the remote resources"""
+
+        raise NotImplementedError("Subclasses must implement this method")
+
+
+    def stop(self):
+        """Stop any threads and prepare for application shutdown"""
+
+        raise NotImplementedError("Subclasses must implement this method")

solarconditionsproviders/giroionosonde.py

@@ -35,6 +35,18 @@ class GIROIonosonde(SolarConditionsProvider):
         self._thread = None
         self._stop_event = Event()
 
+        # Pre-populate ionosonde_data with known station names for stations not already present,
+        # so the station dropdown is available before the first poll. Does not overwrite existing
+        # entries so KC2G cache data is preserved.
+        existing = self._solar_conditions.ionosonde_data or {}
+        new_entries = {
+            s["ursi"]: {"ursi": s["ursi"], "name": s["name"], "fof2": None, "muf": None,
+                        "luf": None, "band_states": None}
+            for s in self._stations if s["ursi"] not in existing
+        }
+        if new_entries:
+            self.update_data({"ionosonde_data": {**existing, **new_entries}})
+
     @staticmethod
     def _load_stations():
         stations = []
@@ -44,21 +56,6 @@ class GIROIonosonde(SolarConditionsProvider):
                     stations.append({"ursi": row[0].strip(), "name": row[1].strip()})
         return stations
 
-    def setup(self, solar_conditions):
-        """Pre-populate ionosonde_data with known station names for stations not already present,
-        so the station dropdown is available before the first poll. Does not overwrite existing
-        entries so KC2G cache data is preserved."""
-
-        super().setup(solar_conditions)
-        existing = solar_conditions.ionosonde_data or {}
-        new_entries = {
-            s["ursi"]: {"ursi": s["ursi"], "name": s["name"], "fof2": None, "muf": None,
-                        "luf": None, "band_states": None}
-            for s in self._stations if s["ursi"] not in existing
-        }
-        if new_entries:
-            self.update_data({"ionosonde_data": {**existing, **new_entries}})
-
     def start(self):
         logging.info(f"Set up query of GIRO ionosonde data API every {POLL_INTERVAL} seconds.")
         self._thread = Thread(target=self._run, daemon=True)

solarconditionsproviders/solar_conditions_provider.py

@@ -2,6 +2,8 @@ from datetime import datetime
 
 import pytz
 
+from core.data_store import DATA_STORE
+
 
 class SolarConditionsProvider:
     """Generic solar conditions provider class. Subclasses of this query individual APIs for space weather and
@@ -14,12 +16,7 @@ class SolarConditionsProvider:
         self.enabled = provider_config["enabled"]
         self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
         self.status = "Not Started" if self.enabled else "Disabled"
-        self._solar_conditions = None
+        self._solar_conditions = DATA_STORE.solar_conditions
-
-    def setup(self, solar_conditions):
-        """Set up the provider, giving it the solar conditions object"""
-
-        self._solar_conditions = solar_conditions
 
     def start(self):
         """Start the provider. This should return immediately after spawning threads to access the remote resources"""

spothole.py

@@ -13,7 +13,6 @@ from core.status_reporter import StatusReporter
 from server.webserver import WebServer
 
 # Globals
-data_store = DATA_STORE
 web_server = None
 spot_providers = []
 alert_providers = []
@@ -39,7 +38,7 @@ def shutdown(_signum=None, _frame=None):
     for scp in solar_condition_providers:
         if scp.enabled:
             scp.stop()
-    data_store.close()
+    DATA_STORE.close()
     os._exit(0)
 
 
@@ -90,13 +89,12 @@ if __name__ == '__main__':
     lookup_helper.start()
 
     # Set up web server
-    web_server = WebServer(data_store=data_store)
+    web_server = WebServer()
 
     # Fetch, set up and start spot providers
     for entry in config["spot-providers"]:
         spot_providers.append(get_spot_provider_from_config(entry))
     for p in spot_providers:
-        p.setup(spots=data_store.spots, web_server=web_server)
         if p.enabled:
             p.start()
 
@@ -104,7 +102,6 @@ if __name__ == '__main__':
     for entry in config["alert-providers"]:
         alert_providers.append(get_alert_provider_from_config(entry))
     for p in alert_providers:
-        p.setup(alerts=data_store.alerts, web_server=web_server)
         if p.enabled:
             p.start()
 
@@ -112,12 +109,11 @@ if __name__ == '__main__':
     for entry in config.get("solar-condition-providers", []):
         solar_condition_providers.append(get_solar_conditions_provider_from_config(entry))
     for p in solar_condition_providers:
-        p.setup(solar_conditions=data_store.solar_conditions)
         if p.enabled:
             p.start()
 
     # Set up status reporter
-    status_reporter = StatusReporter(data_store=data_store, web_server=web_server, spot_providers=spot_providers,
+    status_reporter = StatusReporter(web_server=web_server, spot_providers=spot_providers,
                                      alert_providers=alert_providers,
                                      solar_condition_providers=solar_condition_providers, run_interval=5)
     status_reporter.start()

spotproviders/gma.py

@@ -3,8 +3,8 @@ from datetime import datetime
 
 import pytz
 
-from core.url_data_cache import URL_DATA_CACHE
 from core.constants import HTTP_HEADERS
+from core.url_data_cache import URL_DATA_CACHE
 from data.sig_ref import SIGRef
 from data.spot import Spot
 from spotproviders.http_spot_provider import HTTPSpotProvider

spotproviders/spot_provider.py

@@ -2,7 +2,7 @@ from datetime import datetime
 
 import pytz
 
-from core.config import MAX_SPOT_AGE
+from core.data_store import DATA_STORE
 
 
 class SpotProvider:
@@ -16,14 +16,7 @@ class SpotProvider:
         self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
         self.last_spot_time = datetime.min.replace(tzinfo=pytz.UTC)
         self.status = "Not Started" if self.enabled else "Disabled"
-        self._spots = None
+        self._spots = DATA_STORE.spots
-        self._web_server = None
-
-    def setup(self, spots, web_server):
-        """Set up the provider, e.g. giving it the spot list to work from"""
-
-        self._spots = spots
-        self._web_server = web_server
 
     def start(self):
         """Start the provider. This should return immediately after spawning threads to access the remote resources"""
@@ -60,9 +53,6 @@ class SpotProvider:
     def _add_spot(self, spot):
         if not spot.expired():
             self._spots.set(spot.id, spot)
-            # Ping the web server in case we have any SSE connections that need to see this immediately
-            if self._web_server:
-                self._web_server.notify_new_spot(spot)
 
     def stop(self):
         """Stop any threads and prepare for application shutdown"""