Merge pull request #125 from lpsinger/heartbeat

Add heartbeat topic, `gcn.heartbeat`

gcn_classic_to_kafka/cli.py

@@ -6,6 +6,7 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 """Command line interface."""
+
 import asyncio
 import logging
 import urllib
@@ -17,6 +18,7 @@
 import prometheus_client
 
 from .socket import client_connected
+from . import heartbeat
 from . import metrics
 
 log = logging.getLogger(__name__)
@@ -94,7 +96,10 @@ async def serve():
         async with server:
             await server.serve_forever()
 
+    async def run():
+        await asyncio.gather(asyncio.gather(heartbeat.run(producer), serve()))
+
     # Exit cleanly on SIGTERM
     signal.signal(signal.SIGTERM, signal_handler)
 
-    asyncio.run(serve())
+    asyncio.run(run())

gcn_classic_to_kafka/heartbeat.py

@@ -0,0 +1,39 @@
+#
+# Copyright © 2023 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+import asyncio
+import datetime
+import json
+import logging
+
+import confluent_kafka
+
+log = logging.getLogger(__name__)
+
+TIMEOUT = 1
+TOPIC = "gcn.heartbeat"
+
+
+async def run(producer: confluent_kafka.Producer):
+    """Produce a heartbeat message once per second."""
+    log.info("Producing heartbeats every %d second(s) on topic %s", TIMEOUT, TOPIC)
+    while True:
+        producer.produce(
+            TOPIC,
+            json.dumps(
+                {
+                    "$schema": "https://gcn.nasa.gov/docs/schema/v4.1.0/gcn/notices/core/Alert.schema.json",
+                    "alert_datetime": datetime.datetime.now(datetime.UTC).isoformat(),
+                }
+            ),
+        )
+
+        # Wait for any outstanding messages to be delivered and delivery
+        # report callbacks to be triggered.
+        producer.poll(0)
+
+        await asyncio.sleep(TIMEOUT)