Add output serialization and deserialization support (#6152)

Introduced `Output` message and related data serialization/deserialization methods in `ProtoOutputExtensions`. Updated `WorkflowExecutionResult` and impacted methods to include `Output` handling. Enhanced workflow runtime logic to manage and integrate output data.

src/modules/Elsa.ProtoActor/Extensions/ProtoOutputExtensions.cs

@@ -0,0 +1,15 @@
+using Elsa.ProtoActor.ProtoBuf;
+
+namespace Elsa.ProtoActor.Extensions;
+
+internal static class ProtoOutputExtensions
+{
+    public static IDictionary<string, object> Deserialize(this Output output) => output.Data.Deserialize();
+
+    public static Output SerializeOutput(this IDictionary<string, object> output)
+    {
+        var result = new Output();
+        output.Serialize(result.Data);
+        return result;
+    }
+}

src/modules/Elsa.ProtoActor/Mappers/WorkflowExecutionResultMapper.cs

@@ -42,7 +42,8 @@ internal class WorkflowExecutionResultMapper
             _workflowSubStatusMapper.Map(source.SubStatus),
             _bookmarkMapper.Map(source.Bookmarks).ToList(),
             _activityIncidentStateMapper.Map(source.Incidents).ToList(),
-            source.TriggeredActivityId.NullIfEmpty()
+            source.TriggeredActivityId.NullIfEmpty(),
+            source.Output.Deserialize()
         );
     }
 
@@ -61,6 +62,7 @@ internal class WorkflowExecutionResultMapper
             Bookmarks = { _bookmarkMapper.Map(source.Bookmarks) },
             Incidents = { _activityIncidentStateMapper.Map(source.Incidents).ToList() },
             TriggeredActivityId = source.TriggeredActivityId,
+            Output = source.Output.SerializeOutput()
         };
     }
 }

src/modules/Elsa.ProtoActor/Proto/Shared.proto

@@ -14,6 +14,10 @@ message Input {
   map<string, Json> Data = 1;
 }
 
+message Output {
+  map<string, Json> Data = 1;
+}
+
 message Properties {
   map<string, Json> Data = 1;
 }

src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.Messages.proto

@@ -15,8 +15,8 @@ message StartWorkflowRequest {
   string InstanceId = 2;
   string VersionOptions = 3;
   optional string CorrelationId = 4;
-  optional Input input = 5;
-  optional Properties properties = 6;
+  optional Input Input = 5;
+  optional Properties Properties = 6;
   optional string TriggerActivityId = 7;
   optional bool IsExistingInstance = 8;
 }
@@ -29,6 +29,7 @@ message WorkflowExecutionResponse {
   repeated Bookmark Bookmarks = 5;
   repeated ActivityIncident Incidents = 6;
   optional string TriggeredActivityId = 7;
+  optional Output Output = 8;
 }
 
 message ActivityIncident {

src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs

@@ -5,7 +5,8 @@ using Elsa.Workflows.Attributes;
 using Elsa.Workflows.Contracts;
 using Elsa.Workflows.Management;
 using Elsa.Workflows.Models;
-using Elsa.Workflows.Options;
+using Elsa.Workflows.Runtime.Contracts;
+using Elsa.Workflows.Runtime.Parameters;
 using Elsa.Workflows.UIHints;
 using JetBrains.Annotations;
 
@@ -61,7 +62,7 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
         var workflowDefinitionId = WorkflowDefinitionId.Get(context);
         var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
         var correlationId = CorrelationId.GetOrDefault(context);
-        var workflowInvoker = context.GetRequiredService<IWorkflowRunner>();
+        var workflowRuntime = context.GetRequiredService<IWorkflowRuntime>();
         var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
         var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
         var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken);
@@ -69,21 +70,21 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
         if (workflowGraph == null)
             throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found.");
 
-        var options = new RunWorkflowOptions
+        var options = new StartWorkflowRuntimeParams
         {
             ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id,
             Input = input,
             CorrelationId = correlationId,
-            WorkflowInstanceId = identityGenerator.GenerateId()
+            InstanceId = identityGenerator.GenerateId()
         };
 
-        var workflowResult = await workflowInvoker.RunAsync(workflowGraph, options, context.CancellationToken);
+        var workflowResult = await workflowRuntime.StartWorkflowAsync(workflowDefinitionId, options);
         var info = new ExecuteWorkflowResult
         {
-            WorkflowInstanceId = options.WorkflowInstanceId,
-            Status = workflowResult.WorkflowState.Status,
-            SubStatus = workflowResult.WorkflowState.SubStatus,
-            Output = workflowResult.WorkflowState.Output
+            WorkflowInstanceId = workflowResult.WorkflowInstanceId,
+            Status = workflowResult.Status,
+            SubStatus = workflowResult.SubStatus,
+            Output = workflowResult.Output
         };
 
         return info;

src/modules/Elsa.Workflows.Runtime/Results/WorkflowExecutionResult.cs

@@ -2,4 +2,11 @@ using Elsa.Workflows.Models;
 
 namespace Elsa.Workflows.Runtime.Results;
 
-public record WorkflowExecutionResult(string WorkflowInstanceId, WorkflowStatus Status, WorkflowSubStatus SubStatus, ICollection<Bookmark> Bookmarks, ICollection<ActivityIncident> Incidents, string? TriggeredActivityId = null);
+public record WorkflowExecutionResult(
+    string WorkflowInstanceId, 
+    WorkflowStatus Status, 
+    WorkflowSubStatus SubStatus, 
+    ICollection<Bookmark> Bookmarks, 
+    ICollection<ActivityIncident> Incidents, 
+    string? TriggeredActivityId,
+    IDictionary<string, object> Output);

src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs

@@ -168,7 +168,7 @@ public class DefaultWorkflowRuntime(
             await workflowHost.ResumeWorkflowAsync(resumeWorkflowOptions, applicationCancellationToken);
             await workflowHost.PersistStateAsync(systemCancellationToken);
             workflowState = workflowHost.WorkflowState;
-            return new WorkflowExecutionResult(workflowState.Id, workflowState.Status, workflowState.SubStatus, workflowState.Bookmarks, workflowState.Incidents);
+            return new WorkflowExecutionResult(workflowState.Id, workflowState.Status, workflowState.SubStatus, workflowState.Bookmarks, workflowState.Incidents, null, workflowState.Output);
         }
     }
 
@@ -349,10 +349,10 @@ public class DefaultWorkflowRuntime(
     {
         var workflowInstanceId = string.IsNullOrEmpty(options?.InstanceId)
             ? identityGenerator.GenerateId()
-            : options?.InstanceId;
+            : options.InstanceId;
         var cancellationTokens = options?.CancellationTokens ?? default;
 
-        await using (await AcquireLockAsync(workflowInstanceId!, cancellationTokens.SystemCancellationToken))
+        await using (await AcquireLockAsync(workflowInstanceId, cancellationTokens.SystemCancellationToken))
         {
             var input = options?.Input;
             var correlationId = options?.CorrelationId;
@@ -377,7 +377,8 @@ public class DefaultWorkflowRuntime(
                 workflowState.SubStatus,
                 workflowState.Bookmarks,
                 workflowState.Incidents,
-                default);
+                default,
+                workflowState.Output);
         }
     }
 
@@ -426,8 +427,10 @@ public class DefaultWorkflowRuntime(
                 });
 
             if (resumeResult != null)
-                resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Status,
-                    resumeResult.SubStatus, resumeResult.Bookmarks, resumeResult.Incidents));
+                resumedWorkflows.Add(resumeResult with
+                {
+                    WorkflowInstanceId = workflowInstanceId
+                });
         }
 
         return resumedWorkflows;