mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-08-06 10:31:42 +00:00
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
import logging
|
|
import threading
|
|
import time
|
|
|
|
import diskcache
|
|
from cachetools import TTLCache
|
|
|
|
|
|
class LiveDataCache:
|
|
"""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 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()
|
|
self._start_periodic_snapshot(snapshot_interval_sec)
|
|
|
|
def set(self, key, value):
|
|
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)
|
|
|
|
def delete(self, key):
|
|
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
|
|
data = [(k, v, time.time()) for k, v in self._cache.items()]
|
|
try:
|
|
self._disk_cache.set("snapshot", data)
|
|
except Exception as e:
|
|
logging.error("Failed to write snapshot to %s", self._snapshot_dir, e, exc_info=True)
|
|
|
|
def _load_snapshot(self):
|
|
data = self._disk_cache.get("snapshot")
|
|
if not data:
|
|
return
|
|
|
|
now = time.time()
|
|
with self._lock:
|
|
for key, value, saved_at in data:
|
|
# Only restore entries that would still be within TTL
|
|
if now - saved_at < self._ttl:
|
|
self._cache[key] = value
|
|
logging.info("Loaded snapshot from %s", self._snapshot_dir)
|
|
|
|
def _start_periodic_snapshot(self, interval):
|
|
def loop():
|
|
while True:
|
|
time.sleep(interval)
|
|
self.save_snapshot()
|
|
|
|
t = threading.Thread(target=loop, daemon=True, name=f"snapshot-{self._snapshot_dir}")
|
|
t.start()
|
|
|
|
def close(self):
|
|
self.save_snapshot()
|
|
self._disk_cache.close()
|