Read pending events from database on startup for immediate flushing

Closes #85.
eht16 · Aug 27, 2023 · b9a12a7 · b9a12a7 · Cthulhuisreal · Sep 11, 2023
1 parent aea33fb
commit b9a12a7
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/logstash_async/database.py b/logstash_async/database.py
@@ -158,3 +158,12 @@ def vacuum(self):
         with self._connect() as connection:
             cursor = connection.cursor()
             cursor.execute("VACUUM;")
+
+    # ----------------------------------------------------------------------
+    def get_non_flushed_event_count(self):
+        query_fetch = '''SELECT count(*) FROM `event` WHERE `pending_delete` = 0;'''
+        with self._connect() as connection:
+            cursor = connection.cursor()
+            cursor.execute(query_fetch)
+            count = cursor.fetchone()[0]
+        return count
diff --git a/logstash_async/worker.py b/logstash_async/worker.py
@@ -129,6 +129,8 @@ def _setup_database(self):
         else:
             self._database = MemoryCache(cache=self._memory_cache, event_ttl=self._event_ttl)
 
+        self._non_flushed_event_count = self._database.get_non_flushed_event_count()
+
     # ----------------------------------------------------------------------
     def _fetch_events(self):
         while True: