[WIP] Added TTLCacheDatastore with potential PubSub #1216
Conversation
9356287 to
433e933
Compare
There was a problem hiding this comment.
Pull request overview
This PR introduces a new in-memory TTLCache-based streaming datastore as an alternative to Redis, along with a refactoring that extracts common WebSocket handler logic. The implementation provides a lightweight option for development and testing scenarios where Redis is not available.
Key Changes:
- Added
TTLCacheDatastoreclass usingcachetools.TTLCachefor in-memory sequence and data storage - Introduced
PubSubclass for in-process pub/sub messaging using weakrefs - Refactored common WebSocket handler logic into
_make_ws_handler_commonfunction to reduce code duplication between Redis and TTLCache backends
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def __init__(self, settings: Dict[str, Any]): | ||
| self._settings = settings | ||
| maxsize = self._settings.get("maxsize", 1000) | ||
| seq_ttl = self._settings["seq_ttl"] | ||
| self._seq_cache = cachetools.TTLCache( | ||
| maxsize=maxsize, | ||
| ttl=seq_ttl, | ||
| ) | ||
| self._seq_counters = cachetools.TTLCache( | ||
| maxsize=maxsize, | ||
| ttl=seq_ttl, | ||
| ) | ||
| self._data_cache = cachetools.TTLCache( | ||
| maxsize=maxsize, ttl=self._settings["data_ttl"] | ||
| ) |
There was a problem hiding this comment.
Missing required configuration validation. The TTLCacheDatastore expects seq_ttl and data_ttl settings but doesn't handle the case where they're missing. Unlike RedisStreamingDatastore which requires these settings, TTLCacheDatastore should either provide defaults or validate their presence.
The code will raise a KeyError on line 226 and 236 if these settings are not provided. Consider either:
- Adding validation to raise a clearer error message
- Providing sensible defaults like the
StreamingCacheconfig does (data_ttl=3600, seq_ttl=2592000)
|
The feedback from Copilot is solid. |
Agreed, will start addressing them |
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
…it for TTLCache Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 1 out of 1 changed files in this pull request and generated 3 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
tiled/server/streaming.py
Outdated
| async def publish(self, topic: str, message): | ||
| for ref in list(self._topics.get(topic, ())): | ||
| q = ref() | ||
| if q: | ||
| q.put_nowait(message) |
There was a problem hiding this comment.
The put_nowait call could raise asyncio.QueueFull if a subscriber's queue is full. This would prevent messages from being delivered to other subscribers. Consider wrapping this in a try-except block to log the error and continue delivery to other subscribers.
Checklist