Implement threshold writing to Redis and manticore ingesting from Redis

sources/ingest.py

@@ -0,0 +1,33 @@
+import db
+import util
+import ujson
+import asyncio
+
+SOURCES = ["irc"]
+KEYPREFIX = "queue."
+CHUNK_SIZE = 1000
+ITER_DELAY = 0.5
+
+class Ingest(object):
+    def __init__(self):
+        name = self.__class__.__name__
+        self.log = util.get_logger(name)
+
+    async def run(self):
+        while True:
+            await self.process_chunk()
+            await asyncio.sleep(ITER_DELAY)
+
+    async def process_chunk(self):
+        items = []
+        for source in SOURCES:
+            key = f"{KEYPREFIX}{source}"
+            chunk = await db.ar.spop(key, CHUNK_SIZE)
+            if not chunk:
+                continue
+            self.log.info(f"Got chunk: {chunk}")
+            for item in chunk:
+                item = ujson.loads(item)
+                self.log.info(f"Got item: {item}")
+                items.append(item)
+        db.store_message_bulk(items)