Kafka: Added Serialized Headers to WorkflowInput (#4572)

* Added serialized message headers to MessageReceivedInput --> WorkflowInput * Changed type of headers from string to dictionary<string,string> --------- Co-authored-by: Yannick Laubscher <[email protected]>
elsa-workflows · Oct 30, 2023 · 2caab0c · 2caab0c
1 parent 39c3504
commit 2caab0c
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 5 deletions.
diff --git a/src/activities/Elsa.Activities.Kafka/Elsa.Activities.Kafka.csproj b/src/activities/Elsa.Activities.Kafka/Elsa.Activities.Kafka.csproj
@@ -12,15 +12,15 @@
         <Nullable>enable</Nullable>
     </PropertyGroup>
 
-    <ItemGroup>
-      <ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
-    </ItemGroup>
-
     <ItemGroup>
       <PackageReference Include="Confluent.Kafka" Version="1.9.0" />
       <PackageReference Include="Confluent.SchemaRegistry" Version="1.9.0" />
       <PackageReference Include="Confluent.SchemaRegistry.Serdes.Avro" Version="1.9.0" />
       <PackageReference Include="System.Reactive" Version="5.0.0" />
     </ItemGroup>
 
+    <ItemGroup>
+      <ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
+    </ItemGroup>
+
 </Project>
diff --git a/src/activities/Elsa.Activities.Kafka/Models/MessageReceivedInput.cs b/src/activities/Elsa.Activities.Kafka/Models/MessageReceivedInput.cs
@@ -1,3 +1,4 @@
+using Confluent.Kafka;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -8,6 +9,8 @@ namespace Elsa.Activities.Kafka.Models
 {
     public class MessageReceivedInput
     {
+        public Dictionary<string, string>? MessageHeaders { get; set; }
+
         public string? MessageString { get; set; }
 
         public byte[]? MessageBytes { get; set; }

diff --git a/src/activities/Elsa.Activities.Kafka/Services/Worker.cs b/src/activities/Elsa.Activities.Kafka/Services/Worker.cs
@@ -14,6 +14,7 @@
 using Elsa.Services.Models;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
 
 namespace Elsa.Activities.Kafka.Services
 {
@@ -91,7 +92,7 @@ private async Task TriggerWorkflowsAsync(KafkaMessageEvent ev, CancellationToken
             using var scope = _scopeFactory.CreateScope();
             var config = _client.Configuration;
             var tenantId = await _tenantIdResolver.ResolveAsync(ev, config.Topic, config.Group, Tags, cancellationToken);
-            var workflowInput = new WorkflowInput(new MessageReceivedInput() { MessageBytes = ev.Message.Value, MessageString = Encoding.ASCII.GetString(ev.Message.Value) });
+            var workflowInput = new WorkflowInput(new MessageReceivedInput() { MessageBytes = ev.Message.Value, MessageString = Encoding.ASCII.GetString(ev.Message.Value), MessageHeaders = GetHeaders(ev.Message.Headers,false) });
 
             // Schema extraction if injected
             var schemaResolver = scope.ServiceProvider.GetRequiredService<ISchemaResolver>();