SSE updates every 5 seconds is probably fine, we don't really need every second. #3

alertproviders/alert_provider.py

@@ -30,6 +30,9 @@ class AlertProvider:
     # because alerts could be created at any point for any time in the future. Rely on hashcode-based id matching
     # to deal with duplicates.
     def submit_batch(self, alerts):
+        # Sort the batch so that earliest ones go in first. This helps keep the ordering correct when alerts are fired
+        # off to SSE listeners.
+        alerts = sorted(alerts, key=lambda alert: (alert.start_time if alert and alert.start_time else 0))
         for alert in alerts:
             # Fill in any blanks and add to the list
             alert.infer_missing()

requirements.txt

@@ -13,4 +13,5 @@ pyproj~=3.7.2
 prometheus_client~=0.23.1
 beautifulsoup4~=4.14.2
 websocket-client~=1.9.0
-tornado~=6.5.4
+tornado~=6.5.4
+tornado_eventsource~=3.0.0

server/handlers/api/alerts.py

@@ -1,13 +1,18 @@
 import json
 import logging
 from datetime import datetime
+from queue import Queue
 
 import pytz
 import tornado
+import tornado_eventsource.handler
 
 from core.prometheus_metrics_handler import api_requests_counter
 from core.utils import serialize_everything
 
+SSE_HANDLER_MAX_QUEUE_SIZE = 100
+SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000
+
 
 # API request handler for /api/v1/alerts
 class APIAlertsHandler(tornado.web.RequestHandler):
@@ -43,28 +48,61 @@ class APIAlertsHandler(tornado.web.RequestHandler):
         self.set_header("Content-Type", "application/json")
 
 # API request handler for /api/v1/alerts/stream
-class APIAlertsStreamHandler(tornado.web.RequestHandler):
+class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
-    def get(self):
+    def initialize(self, sse_alert_queues, web_server_metrics):
-        # todo
+        self.sse_alert_queues = sse_alert_queues
-        # try:
+        self.web_server_metrics = web_server_metrics
-        #     # Metrics
+
-        #     api_requests_counter.inc()
+    def open(self):
-        #
+        try:
-        #     response.content_type  = 'text/event-stream'
+            # Metrics
-        #     response.cache_control = 'no-cache'
+            self.web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
-        #     yield 'retry: 1000\n\n'
+            self.web_server_metrics["api_access_counter"] += 1
-        #
+            self.web_server_metrics["status"] = "OK"
-        #     alert_queue = Queue(maxsize=100)
+            api_requests_counter.inc()
-        #     self.sse_alert_queues.append(alert_queue)
+
-        #     while True:
+            # request.arguments contains lists for each param key because technically the client can supply multiple,
-        #         if alert_queue.empty():
+            # reduce that to just the first entry, and convert bytes to string
-        #             gevent.sleep(1)
+            self.query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
-        #         else:
+
-        #             alert = alert_queue.get()
+            # Create a alert queue and add it to the web server's list. The web server will fill this when alerts arrive
-        #             yield 'data: ' + json.dumps(alert, default=serialize_everything) + '\n\n'
+            self.alert_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE)
-        # except Exception as e:
+            self.sse_alert_queues.append(self.alert_queue)
-        #     logging.warn("Exception when serving SSE socket", e)
+
-        pass
+            # 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()
+
+        except Exception as e:
+            logging.warn("Exception when serving SSE socket", e)
+
+    # When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it
+    def close(self):
+        try:
+            if self.alert_queue in self.sse_alert_queues:
+                self.sse_alert_queues.remove(self.alert_queue)
+            self.alert_queue.empty()
+        except:
+            pass
+        self.alert_queue = None
+        super().close()
+
+    # Callback to check if anything has arrived in the queue, and if so send it to the client
+    def _callback(self):
+        try:
+            if self.alert_queue:
+                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):
+                        self.write_message(msg=json.dumps(alert, default=serialize_everything))
+
+                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:
+            logging.warn("Exception in SSE callback, connection will be closed.")
+            self.close()
 
 
 
@@ -127,4 +165,10 @@ def alert_allowed_by_query(alert, query):
                 dx_call_includes = query.get(k).strip()
                 if not alert.dx_call or dx_call_includes.upper() not in alert.dx_call.upper():
                     return False
+            case "text_includes":
+                text_includes = query.get(k).strip()
+                if (not alert.dx_call or text_includes.upper() not in alert.dx_call.upper()) \
+                        and (not alert.comment or text_includes.upper() not in alert.comment.upper()) \
+                        and (not alert.freqs_modes or text_includes.upper() not in alert.freqs_modes.upper()):
+                    return False
     return True

server/handlers/api/spots.py

@@ -1,13 +1,18 @@
 import json
 import logging
 from datetime import datetime, timedelta
+from queue import Queue
 
 import pytz
 import tornado
+import tornado_eventsource.handler
 
 from core.prometheus_metrics_handler import api_requests_counter
 from core.utils import serialize_everything
 
+SSE_HANDLER_MAX_QUEUE_SIZE = 1000
+SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000
+
 
 # API request handler for /api/v1/spots
 class APISpotsHandler(tornado.web.RequestHandler):
@@ -44,28 +49,62 @@ class APISpotsHandler(tornado.web.RequestHandler):
 
 
 # API request handler for /api/v1/spots/stream
-class APISpotsStreamHandler(tornado.web.RequestHandler):
+class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
-    def get(self):
+    def initialize(self, sse_spot_queues, web_server_metrics):
-        # todo
+        self.sse_spot_queues = sse_spot_queues
-        # try:
+        self.web_server_metrics = web_server_metrics
-        #     # Metrics
+
-        #     api_requests_counter.inc()
+    # Called once on the client opening a connection, set things up
-        #
+    def open(self):
-        #     response.content_type  = 'text/event-stream'
+        try:
-        #     response.cache_control = 'no-cache'
+            # Metrics
-        #     yield 'retry: 1000\n\n'
+            self.web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
-        #
+            self.web_server_metrics["api_access_counter"] += 1
-        #     spot_queue = Queue(maxsize=100)
+            self.web_server_metrics["status"] = "OK"
-        #     self.sse_spot_queues.append(spot_queue)
+            api_requests_counter.inc()
-        #     while True:
+
-        #         if spot_queue.empty():
+            # request.arguments contains lists for each param key because technically the client can supply multiple,
-        #             gevent.sleep(1)
+            # reduce that to just the first entry, and convert bytes to string
-        #         else:
+            self.query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
-        #             spot = spot_queue.get()
+
-        #             yield 'data: ' + json.dumps(spot, default=serialize_everything) + '\n\n'
+            # Create a spot queue and add it to the web server's list. The web server will fill this when spots arrive
-        # except Exception as e:
+            self.spot_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE)
-        #     logging.warn("Exception when serving SSE socket", e)
+            self.sse_spot_queues.append(self.spot_queue)
-        pass
+
+            # 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()
+
+        except Exception as e:
+            logging.warn("Exception when serving SSE socket", e)
+
+    # When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it
+    def close(self):
+        try:
+            if self.spot_queue in self.sse_spot_queues:
+                self.sse_spot_queues.remove(self.spot_queue)
+            self.spot_queue.empty()
+        except:
+            pass
+        self.spot_queue = None
+        super().close()
+
+    # Callback to check if anything has arrived in the queue, and if so send it to the client
+    def _callback(self):
+        try:
+            if self.spot_queue:
+                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):
+                        self.write_message(msg=json.dumps(spot, default=serialize_everything))
+
+                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:
+            logging.warn("Exception in SSE callback, connection will be closed.")
+            self.close()
 
 
 
@@ -177,6 +216,11 @@ def spot_allowed_by_query(spot, query):
                 dx_call_includes = query.get(k).strip()
                 if not spot.dx_call or dx_call_includes.upper() not in spot.dx_call.upper():
                     return False
+            case "text_includes":
+                text_includes = query.get(k).strip()
+                if (not spot.dx_call or text_includes.upper() not in spot.dx_call.upper()) \
+                        and (not spot.comment or text_includes.upper() not in spot.comment.upper()):
+                    return False
             case "allow_qrt":
                 # If false, spots that are flagged as QRT are not returned.
                 prevent_qrt = query.get(k).upper() == "FALSE"

server/webserver.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import os
 
 import tornado
@@ -15,8 +16,6 @@ from server.handlers.pagetemplate import PageTemplateHandler
 
 
 # Provides the public-facing web server.
-# TODO SSE API responses
-# TODO clean_up_sse_queues
 class WebServer:
     # Constructor
     def __init__(self, spots, alerts, status_data, port):
@@ -49,8 +48,8 @@ class WebServer:
             # 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), # todo provide queues?
+            (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), # todo provide queues?
+            (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}),
@@ -71,37 +70,51 @@ class WebServer:
             (r"/(.*)", StaticFileHandler, {"path": os.path.join(os.path.dirname(__file__), "../webassets")}),
         ],
             template_path=os.path.join(os.path.dirname(__file__), "../templates"),
-            debug=True) # todo set false
+            debug=False)
         app.listen(self.port)
         await self.shutdown_event.wait()
 
     # 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.
     def notify_new_spot(self, spot):
-        # todo
+        for queue in self.sse_spot_queues:
-        # for queue in self.sse_spot_queues:
+            try:
-        #     try:
+                queue.put(spot)
-        #         queue.put(spot)
+            except:
-        #     except:
+                # Cleanup thread was probably deleting the queue, that's fine
-        #         # Cleanup thread was probably deleting the queue, that's fine
+                pass
-        #         pass
         pass
 
     # 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.
     def notify_new_alert(self, alert):
-        # todo
+        for queue in self.sse_alert_queues:
-        # for queue in self.sse_alert_queues:
+            try:
-        #     try:
+                queue.put(alert)
-        #         queue.put(alert)
+            except:
-        #     except:
+                # Cleanup thread was probably deleting the queue, that's fine
-        #         # Cleanup thread was probably deleting the queue, that's fine
+                pass
-        #         pass
         pass
 
-    # Clean up any SSE queues that are growing too large; probably their client disconnected.
+    # Clean up any SSE queues that are growing too large; probably their client disconnected and we didn't catch it
+    # properly for some reason.
     def clean_up_sse_queues(self):
-        # todo
+        for q in self.sse_spot_queues:
-        # self.sse_spot_queues = [q for q in self.sse_spot_queues if not q.full()]
+            try:
-        # self.sse_alert_queues = [q for q in self.sse_alert_queues if not q.full()]
+                if q.full():
+                    logging.warn("A full SSE spot queue was found, presumably because the client disconnected strangely. It has been removed.")
+                    self.sse_spot_queues.remove(q)
+                    q.empty()
+            except:
+                # Probably got deleted already on another thread
+                pass
+        for q in self.sse_alert_queues:
+            try:
+                if q.full():
+                    logging.warn("A full SSE alert queue was found, presumably because the client disconnected strangely. It has been removed.")
+                    self.sse_alert_queues.remove(q)
+                    q.empty()
+            except:
+                # Probably got deleted already on another thread
+                pass
         pass

spotproviders/spot_provider.py

@@ -32,6 +32,9 @@ class SpotProvider:
     # their infer_missing() method called to complete their data set. This is called by the API-querying
     # subclasses on receiving spots.
     def submit_batch(self, spots):
+        # Sort the batch so that earliest ones go in first. This helps keep the ordering correct when spots are fired
+        # off to SSE listeners.
+        spots = sorted(spots, key=lambda spot: (spot.time if spot and spot.time else 0))
         for spot in spots:
             if datetime.fromtimestamp(spot.time, pytz.UTC) > self.last_spot_time:
                 # Fill in any blanks and add to the list

spotproviders/websocket_spot_provider.py

@@ -47,6 +47,7 @@ class WebsocketSpotProvider(SpotProvider):
                 logging.debug("Connecting to " + self.name + " spot API...")
                 self.status = "Connecting"
                 self.ws = create_connection(self.url, header=HTTP_HEADERS)
+                self.status = "Connected"
                 data = self.ws.recv()
                 if data:
                     try:

templates/base.html

@@ -63,7 +63,7 @@
                     <li class="nav-item ms-4"><a href="/bands" class="nav-link" id="nav-link-bands"><i class="fa-solid fa-ruler-vertical"></i> Bands</a></li>
                     <li class="nav-item ms-4"><a href="/alerts" class="nav-link" id="nav-link-alerts"><i class="fa-solid fa-bell"></i> Alerts</a></li>
                   {% if allow_spotting %}
-                    <li class="nav-item ms-4"><a href="/add-spot" class="nav-link" id="nav-link-add-spot"><i class="fa-solid fa-comment"></i> Add Spot</a></li>
+                    <li class="nav-item ms-4"><a href="/add-spot" class="nav-link" id="nav-link-add-spot"><i class="fa-solid fa-comment"></i> Add&nbsp;Spot</a></li>
                   {% end %}
                     <li class="nav-item ms-4"><a href="/status" class="nav-link" id="nav-link-status"><i class="fa-solid fa-chart-simple"></i> Status</a></li>
                     <li class="nav-item ms-4"><a href="/about" class="nav-link" id="nav-link-about"><i class="fa-solid fa-circle-info"></i> About</a></li>

templates/spots.html

@@ -10,14 +10,21 @@
 
 <div class="mt-3">
     <div id="settingsButtonRow" class="row">
-        <div class="col-auto me-auto pt-3">
+        <div class="col-lg-6 me-auto pt-3 hideonmobile">
             <p id="timing-container">Loading...</p>
         </div>
-        <div class="col-auto">
+        <div class="col-lg-6 text-end">
             <p class="d-inline-flex gap-1">
-                <span style="position: relative;">
+                <span class="btn-group" role="group">
-                    <i class="fa-solid fa-magnifying-glass" style="position: absolute; left: 0px; top: 2px; padding: 10px; pointer-events: none;"></i>
+                  <input type="radio" class="btn-check" name="runPause" id="runButton" autocomplete="off" checked>
-                    <input id="filter-dx-call" type="search" class="form-control" oninput="filtersUpdated();" placeholder="Callsign">
+                  <label class="btn btn-outline-primary" for="runButton"><i class="fa-solid fa-play"></i> Run</label>
+
+                  <input type="radio" class="btn-check" name="runPause" id="pauseButton" autocomplete="off">
+                  <label class="btn btn-outline-primary" for="pauseButton"><i class="fa-solid fa-pause"></i> Pause</label>
+                </span>
+                <span class="hideonmobile" style="position: relative;">
+                    <i id="searchicon" class="fa-solid fa-magnifying-glass"></i>
+                    <input id="search" type="search" class="form-control" oninput="filtersUpdated();" placeholder="Search">
                 </span>
                 <button id="filters-button" type="button" class="btn btn-outline-primary" data-bs-toggle="button" onclick="toggleFiltersPanel();"><i class="fa-solid fa-filter"></i> Filters</button>
                 <button id="display-button" type="button" class="btn btn-outline-primary" data-bs-toggle="button" onclick="toggleDisplayPanel();"><i class="fa-solid fa-desktop"></i> Display</button>

webassets/apidocs/openapi.yml

@@ -115,7 +115,7 @@ paths:
             default: false
         - name: dx_call_includes
           in: query
-          description: "Limit the alerts to only ones where the DX callsign includes the supplied string (case-insensitive). Generally a complete callsign, but you can supply a shorter string for partial matches."
+          description: "Limit the spots to only ones where the DX callsign includes the supplied string (case-insensitive). Generally a complete callsign, but you can supply a shorter string for partial matches."
           required: false
           schema:
             type: string
@@ -125,6 +125,12 @@ paths:
           required: false
           schema:
             type: string
+        - name: text_includes
+          in: query
+          description: "Limit the spots to only ones where some significant text (DX callsign or comment) includes the supplied string (case-insensitive)."
+          required: false
+          schema:
+            type: string
         - name: needs_good_location
           in: query
           description: "Return only spots with a 'good' location. (See the spot `dx_location_good` parameter for details. Useful for map-based clients, to avoid spots with 'bad' locations e.g. loads of cluster spots ending up in the centre of the DXCC entitity.)"
@@ -215,7 +221,7 @@ paths:
             $ref: "#/components/schemas/Continent"
         - name: dx_call_includes
           in: query
-          description: "Limit the alerts to only ones where the DX callsign includes the supplied string (case-insensitive). Generally a complete callsign, but you can supply a shorter string for partial matches."
+          description: "Limit the spots to only ones where the DX callsign includes the supplied string (case-insensitive). Generally a complete callsign, but you can supply a shorter string for partial matches."
           required: false
           schema:
             type: string
@@ -225,6 +231,12 @@ paths:
           required: false
           schema:
             type: string
+        - name: text_includes
+          in: query
+          description: "Limit the spots to only ones where some significant text (DX callsign or comment) includes the supplied string (case-insensitive)."
+          required: false
+          schema:
+            type: string
         - name: needs_good_location
           in: query
           description: "Return only spots with a 'good' location. (See the spot `dx_location_good` parameter for details. Useful for map-based clients, to avoid spots with 'bad' locations e.g. loads of cluster spots ending up in the centre of the DXCC entitity.)"
@@ -304,6 +316,12 @@ paths:
           required: false
           schema:
             type: string
+        - name: text_includes
+          in: query
+          description: "Limit the alerts to only ones where some significant text (DX callsign, freqs/modes, or comment) includes the supplied string (case-insensitive)."
+          required: false
+          schema:
+            type: string
       responses:
         '200':
           description: Success
@@ -359,6 +377,12 @@ paths:
           required: false
           schema:
             type: string
+        - name: text_includes
+          in: query
+          description: "Limit the alerts to only ones where some significant text (DX callsign, freqs/modes, or comment) includes the supplied string (case-insensitive)."
+          required: false
+          schema:
+            type: string
       responses:
         '200':
           description: Success

webassets/css/style.css

@@ -80,17 +80,22 @@ div.container {
 
 /* SPOTS/ALERTS PAGES, SETTINGS/STATUS AREAS */
 
-input#filter-dx-call {
+input#search {
     max-width: 12em;
+    margin-left: 1rem;
     margin-right: 1rem;
     padding-left: 2em;
 }
 
-div.appearing-panel {
+i#searchicon {
-    display: none;
+    position: absolute;
+    left: 1rem;
+    top: 2px;
+    padding: 10px;
+    pointer-events: none;
 }
 
-button#add-spot-button {
+div.appearing-panel {
     display: none;
 }
 
@@ -340,11 +345,6 @@ div.band-spot:hover span.band-spot-info {
         max-height: 26em;
         overflow: scroll;
     }
-    /* Filter/search DX Call field should be smaller on mobile */
-    input#filter-dx-call {
-        max-width: 9em;
-        margin-right: 0;
-    }
 }
 
 @media (min-width: 992px) {

webassets/js/common.js

@@ -117,23 +117,19 @@ function toggleFilterButtons(filterQuery, state) {
 function updateRefreshDisplay() {
     if (lastUpdateTime != null) {
         let secSinceUpdate = moment.duration(moment().diff(lastUpdateTime)).asSeconds();
-        if (typeof REFRESH_INTERVAL_SEC !== 'undefined' && REFRESH_INTERVAL_SEC != null) {
+        let count = REFRESH_INTERVAL_SEC;
-            let count = REFRESH_INTERVAL_SEC;
+        let updatingString = "Updating..."
-            let updatingString = "Updating..."
+        if (secSinceUpdate < REFRESH_INTERVAL_SEC) {
-            if (secSinceUpdate < REFRESH_INTERVAL_SEC) {
+            count = REFRESH_INTERVAL_SEC - secSinceUpdate;
-                count = REFRESH_INTERVAL_SEC - secSinceUpdate;
+            if (count <= 60) {
-                if (count <= 60) {
+                var number = count.toFixed(0);
-                    var number = count.toFixed(0);
+                updatingString = "<span class='nowrap'>Updating in " + number + " second" + (number != "1" ? "s" : "") + ".</span>";
-                    updatingString = "<span class='nowrap'>Updating in " + number + " second" + (number != "1" ? "s" : "") + ".</span>";
+            } else {
-                } else {
+                var number = Math.round(count / 60.0).toFixed(0);
-                    var number = Math.round(count / 60.0).toFixed(0);
+                updatingString = "<span class='nowrap'>Updating in " + number + " minute" + (number != "1" ? "s" : "") + ".</span>";
-                    updatingString = "<span class='nowrap'>Updating in " + number + " minute" + (number != "1" ? "s" : "") + ".</span>";
-                }
             }
-            $("#timing-container").html("Last updated at " + lastUpdateTime.format('HH:mm') + " UTC. " + updatingString);
-        } else {
-            $("#timing-container").html("Connected to live spot server. Last spot at " + lastUpdateTime.format('HH:mm') + " UTC.");
         }
+        $("#timing-container").html("Last updated at " + lastUpdateTime.format('HH:mm') + " UTC. " + updatingString);
     }
 }
 

webassets/js/spots.js

@@ -5,31 +5,37 @@ let rowCount = 0;
 
 // Load spots and populate the table.
 function loadSpots() {
+    // If we have an ongoing SSE connection, stop it so it doesn't interfere with our reload
+    if (evtSource != null) {
+        evtSource.close();
+    }
+
+    // Make the new query
     $.getJSON('/api/v1/spots' + buildQueryString(), function(jsonData) {
         // Store last updated time
         lastUpdateTime = moment.utc();
-        updateRefreshDisplay();
+        updateTimingDisplayRunPause();
         // Store data
         spots = jsonData;
         // Update table
         updateTable();
-        // Start SSE connection to fetch updates in the background
+        // Start SSE connection to fetch updates in the background, if we are in "run" mode
-        restartSSEConnection();
+        let run = $('#runButton:checked').val();
+        if (run) {
+            startSSEConnection();
+        }
     });
 }
 
 // Start an SSE connection (closing an existing one if it exists). This will then be used to add to the table on the
 // fly.
-function restartSSEConnection() {
+function startSSEConnection() {
-    if (evtSource != null) {
+    evtSource = new EventSource('/api/v1/spots/stream' + buildQueryString());
-        evtSource.close();
-    }
-    evtSource = new EventSource('/api/v1/spots/stream');
 
     evtSource.onmessage = function(event) {
         // Store last updated time
         lastUpdateTime = moment.utc();
-        updateRefreshDisplay();
+        updateTimingDisplayRunPause();
         // Get the new spot
         newSpot = JSON.parse(event.data);
         // Awful fudge to ensure new incoming spots at the top of the list don't have timestamps that make them look
@@ -42,17 +48,35 @@ function restartSSEConnection() {
 
         // Add spot to internal data store
         spots.unshift(newSpot);
-        spots = spots.slice(0, -1);
+        // Work out if we need to remove an old spot
-        // Add spot to table
+        if (spots.length > $("#spots-to-fetch option:selected").val()) {
+            spots = spots.slice(0, -1);
+            // Drop oldest spot off the end of the table. This is two rows because of the mobile view extra rows
+            $("#table tbody tr").last().remove();
+            $("#table tbody tr").last().remove();
+        }
+        // If we had zero spots before (i.e. one now), the table will have a "No spots" row that we need to remove now
+        // that we have one.
+        if (spots.length == 1) {
+            $("#table tbody tr").last().remove();
+        }
+
+        // Add the new spot to table
         addSpotToTopOfTable(newSpot, true);
     };
 
     evtSource.onerror = function(err) {
         evtSource.close();
-        setTimeout(restartSSEConnection, 1000);
+        setTimeout(startSSEConnection, 1000);
     };
 }
 
+// Update the special timing display for the live spots page, which varies depending on run/pause selection.
+function updateTimingDisplayRunPause() {
+    let run = $('#runButton:checked').val();
+    $("#timing-container").html((run ? "Connected to server. Last update at " : "Paused at ") + lastUpdateTime.format('HH:mm') + " UTC.");
+}
+
 // Build a query string for the API, based on the filters that the user has selected.
 function buildQueryString() {
     var str =  "?";
@@ -62,8 +86,8 @@ function buildQueryString() {
         }
     });
     str = str + "limit=" + $("#spots-to-fetch option:selected").val();
-    if ($("#filter-dx-call").val() != "") {
+    if ($("#search").val() != "") {
-        str = str + "&dx_call_includes=" + encodeURIComponent($("#filter-dx-call").val());
+        str = str + "&text_includes=" + encodeURIComponent($("#search").val());
     }
     return str;
 }
@@ -216,9 +240,8 @@ function createNewTableRowsForSpot(s, highlightNew) {
     // Format the mode
     mode_string = s["mode"];
     if (s["mode"] == null) {
-        mode_string = "???";
+        mode_string = "";
-    }
+    } else if (s["mode_source"] == "BANDPLAN") {
-    if (s["mode_source"] == "BANDPLAN") {
         mode_string = mode_string + "<span class='mode-q hideonmobile'><i class='fa-solid fa-circle-question' title='The mode was not reported via the spotting service. This is a guess based on the frequency.'></i></span>";
     }
 
@@ -385,11 +408,6 @@ function loadOptions() {
             $("#tableShowBearing").prop('checked', false);
         }
 
-        // Show the Add Spot button if spotting is allowed
-        if (options["spot_allowed"]) {
-            $("#add-spot-button").show();
-        }
-
         // Load spots (this will also set up the SSE connection to update them too)
         loadSpots();
     });
@@ -459,4 +477,19 @@ $(document).ready(function() {
     loadOptions();
     // Display intro box
     displayIntroBox();
+
+    // Set up run/pause toggles
+    $("#runButton").change(function() {
+        // Need to start the SSE connection but also do a full re-query to catch up anything that we missed, so we
+        // might as well just call loadSpots again which will trigger it all
+        loadSpots();
+        updateTimingDisplayRunPause();
+    });
+    $("#pauseButton").change(function() {
+        // If we are pausing and have an open SSE connection, stop it
+        if (evtSource != null) {
+            evtSource.close();
+        }
+        updateTimingDisplayRunPause();
+    });
 });