Merge pull request #1173 from mursisoy/bugfix/sqlite-stale-data

bugfix/sqlite stale data
thingsboard · Sep 12, 2023 · b81fbfa · b81fbfa
2 parents 0e3996b + 3f487d6
commit b81fbfa
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 10 deletions.
diff --git a/thingsboard_gateway/storage/sqlite/database.py b/thingsboard_gateway/storage/sqlite/database.py
@@ -104,17 +104,17 @@ def process(self):
             self.db.rollback()
             log.exception(e)
 
-    def read_data(self, ts):
+    def read_data(self):
         try:
-            data = self.db.execute('''SELECT message FROM messages WHERE timestamp >= ? ;''', [ts])
+            data = self.db.execute('''SELECT timestamp, message FROM messages ORDER BY timestamp ASC LIMIT 0, 50;''')
             return data
         except Exception as e:
             self.db.rollback()
             log.exception(e)
 
     def delete_data(self, ts):
         try:
-            data = self.db.execute('''DELETE FROM messages WHERE timestamp >= ? ;''', [ts])
+            data = self.db.execute('''DELETE FROM messages WHERE timestamp <= ?;''', [ts,])
             return data
         except Exception as e:
             self.db.rollback()

diff --git a/thingsboard_gateway/storage/sqlite/sqlite_event_storage.py b/thingsboard_gateway/storage/sqlite/sqlite_event_storage.py
@@ -46,21 +46,23 @@ def __init__(self, config):
 
     def get_event_pack(self):
         if not self.stopped:
-            self.delete_time_point = self.last_read
-            data_from_storage = self.read_data(self.last_read)
-            self.last_read = time()
-
-            return [item[0] for item in data_from_storage or []]
+            data_from_storage = self.read_data()
+            try:                
+                event_pack_timestamps, event_pack_messages = zip(*([(item[0],item[1]) for item in data_from_storage]))
+            except ValueError as e:
+                return []
+            self.delete_time_point = max(event_pack_timestamps)            
+            return event_pack_messages
         else:
             return []
 
     def event_pack_processing_done(self):
         if not self.stopped:
             self.delete_data(self.delete_time_point)
 
-    def read_data(self, ts):
+    def read_data(self):
         self.db.__stopped = True
-        data = self.db.read_data(ts)
+        data = self.db.read_data()
         self.db.__stopped = False
         return data