place headers under meta

Signed-off-by: clyang82 <[email protected]>
ansible · May 31, 2024 · a4c8742 · a4c8742
1 parent 74894f0
commit a4c8742
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 3 deletions.
diff --git a/extensions/eda/plugins/event_source/kafka.py b/extensions/eda/plugins/event_source/kafka.py
@@ -129,7 +129,8 @@ async def receive_msg(
         # Process headers
         try:
             headers = {header[0]: header[1].decode(encoding) for header in msg.headers}
-            event["headers"] = headers
+            event["meta"] = {}
+            event["meta"]["headers"] = headers
         except UnicodeError:
             logger.exception("Unicode error while decoding headers")
 

diff --git a/tests/integration/event_source_kafka/test_kafka_rules_headers.yml b/tests/integration/event_source_kafka/test_kafka_rules_headers.yml
@@ -9,7 +9,7 @@
         encoding: ascii
   rules:
     - name: match kafka event
-      condition: event.headers.foo == "bar"
+      condition: event.meta.headers.foo == "bar"
       action:
         debug:
           msg: "Rule fired successfully with headers"

diff --git a/tests/unit/event_source/test_kafka.py b/tests/unit/event_source/test_kafka.py
@@ -59,5 +59,8 @@ def test_receive_from_kafka_place_in_queue(myqueue):
                 },
             )
         )
-        assert myqueue.queue[0] == {"body": {"i": 0}, "headers": {"foo": "bar"}}
+        assert myqueue.queue[0] == {
+            "body": {"i": 0},
+            "meta": {"headers": {"foo": "bar"}},
+        }
         assert len(myqueue.queue) == 2