Pull-based indexer

Author: heussd

.devcontainer/python/devcontainer.json

@@ -37,7 +37,7 @@
         }
     },
     "containerEnv": {
-        "GOROOT2": "./python"
+        "NATS_URL": "http://nats-server:4222"
     },
     "shutdownAction": "stopCompose"
 }

python/nats-indexer.py

@@ -1,4 +1,5 @@
 import json
+from typing import List
 import nats
 import asyncio
 
@@ -7,34 +8,42 @@
 import config
 
 
-async def process(msg):
-    print(f"Received a message")
-
-    json_data = json.loads(msg.data.decode())
-    prepared_search_docs = prepare(SearchDoc(
-        title=json_data["title"],
-        author=json_data["author"],
-        content=json_data["content"],
-        excerpt=json_data["excerpt"],
-        date=json_data["date"],
-        language=json_data["language"],
-        url=json_data["url"],
-    ))
-
+async def process(msgs: List[nats.aio.msg.Msg]):
+    jsonmsgs = []
+    for msg in msgs:
+        json_data = json.loads(msg.data.decode())
+        jsonmsgs.append(SearchDoc(
+            title=json_data["title"],
+            author=json_data["author"],
+            content=json_data["content"],
+            excerpt=json_data["excerpt"],
+            date=json_data["date"],
+            language=json_data["language"],
+            url=json_data["url"],
+        ))
+    prepared_search_docs = prepare(jsonmsgs)
     add(prepared_search_docs)
 
 
 async def run():
     nc = await nats.connect(config.NATS_URL)
     js = nc.jetstream()
 
-    await js.subscribe(
-        config.NATS_STREAM_NAME,
-        config=nats.js.api.ConsumerConfig(durable_name=config.NATS_CONSUMER_NAME,
-                                          ack_policy=nats.js.api.AckPolicy.NONE),
-        cb=process,
-        manual_ack=False
-    )
+    psub = await js.pull_subscribe("*",
+                                   stream=config.NATS_STREAM_NAME,
+                                   durable=config.NATS_CONSUMER_NAME)
+
+    while True:
+        try:
+            msgs = await psub.fetch(10, timeout=5)
+            print(f"Fetched {len(msgs)} messages")
+            await process(msgs)
+            for msg in msgs:
+                await msg.ack()
+        except asyncio.TimeoutError:
+            print("No new messages, waiting...")
+            continue
+
 
 
 if __name__ == "__main__":