Merge pull request #336 from neuroglia-io/fix-eventing

Fixed production of cloud events

Author: cdavernas

src/apps/Synapse.Server/appsettings.json

@@ -14,7 +14,7 @@
   },
   "CloudEvents": {
     "Sink": {
-      "Uri": "https://webhook.site/2e78793f-69a5-4ef6-8fdf-d467a9d6f1c6"
+      "Uri": "https://webhook.site/70e57ac3-c28c-41cc-9a61-070810da8b3b"
     }
   }
 }

src/apps/Synapse.Worker/Services/Processors/AsyncFunctionProcessor.cs

@@ -102,6 +102,8 @@ protected override async Task InitializeAsync(CancellationToken cancellationToke
             if (this.Activity.Status == V1WorkflowActivityStatus.Pending)
             {
                 var input = await this.Context.FilterInputAsync(this.Action, this.Activity.Input.ToObject()!, cancellationToken);
+                if (!string.IsNullOrWhiteSpace(this.Action.Event!.DataExpression)) input = await this.Context.EvaluateAsync(this.Action.Event!.DataExpression, input, cancellationToken);
+                else if (this.Action.Event.Data != null) input = await this.Context.EvaluateObjectAsync(this.Action.Event!.Data.ToObject()!, input!, cancellationToken);
                 var metadata = new Dictionary<string, string>() 
                 {
                     {  V1WorkflowActivityMetadata.State, this.State.Name },

src/core/Synapse.Application/Extensions/V1EventExtensions.cs

@@ -41,9 +41,12 @@ public static CloudEvent AsCloudEvent(this V1Event e)
             ce.DataContentType = e.DataContentType;
             ce.DataSchema = e.DataSchema;
             ce.Data = e.Data;
-            foreach(var extension in e.ExtensionAttributes)
+            if(e.ExtensionAttributes != null)
             {
-                ce.SetAttributeFromString(extension.Key, extension.Value?.ToString()!);
+                foreach (var extension in e.ExtensionAttributes)
+                {
+                    ce.SetAttributeFromString(extension.Key, extension.Value?.ToString()!);
+                }
             }
             return ce;
         }