Duckdb draft version

Author: attdona

.gitignore

@@ -128,7 +128,7 @@ celerybeat.pid
 
 # Environments
 .env
-penv/
+.env.test
 
 # Spyder project settings
 .spyderproject

src/rembus/core.py

@@ -342,6 +342,8 @@ def __init__(
         self.owners = rs.load_tenants(self)
         self.start_ts = time.time()
         self.msg_cache: list[rp.PubSubMsg] = []
+        self.msg_topic_cache: dict[str, List[rp.PubSubMsg]] = {}
+        self.tables: dict[str, rdb.Table] = {}
         self._builtins()
         if data_at_rest:
             self.db = rdb.init_db(self, schema)
@@ -423,8 +425,9 @@ async def _pubsub_msg(self, msg: rp.PubSubMsg):
 
         data = rp.tag2df(msg.data)
 
-        # save the message into msg_cache
-        self.msg_cache.append(msg)
+        if self.db is not None:
+            # save the message into msg_cache
+            self.append_message(msg)
 
         try:
             if msg.topic in self.handler:
@@ -441,6 +444,16 @@ async def _pubsub_msg(self, msg: rp.PubSubMsg):
             traceback.print_exc()
         return
 
+    def append_message(self, msg: rp.PubSubMsg):
+        """Append a message to the message cache."""
+        topic = msg.topic
+        rdb.msg_table(self, msg)
+        self.msg_cache.append(msg)
+        if msg.table in self.tables:
+            if topic not in self.msg_topic_cache:
+                self.msg_topic_cache[msg.table] = []
+            self.msg_topic_cache[msg.table].append(msg)
+
     async def send_message(self, msg):
         """Send message to remote node using a twin from the pool of twins"""
         for t in self.id_twin.values():
@@ -763,7 +776,7 @@ def __lt__(self, other):
     def db(self):
         """Return the database associated with this twin."""
         return self.router.db
-    
+
     @property
     def rid(self):
         """Return the unique id of the rembus component."""