Improvements for eryph (#5)

* changes for eryph: passing more data * added serialization config
dbosoft · Jul 28, 2023 · 9a80fe6 · 9a80fe6
1 parent 03bde2c
commit 9a80fe6
Show file tree

Hide file tree

Showing 9 changed files with 37 additions and 17 deletions.
diff --git a/src/Rebus.Operations/Rebus.Operations.Abstractions/Workflow/IOperationManager.cs b/src/Rebus.Operations/Rebus.Operations.Abstractions/Workflow/IOperationManager.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Dbosoft.Rebus.Operations.Workflow;
@@ -7,8 +8,11 @@ namespace Dbosoft.Rebus.Operations.Workflow;
 public interface IOperationManager
 {
     ValueTask<IOperation?> GetByIdAsync(Guid operationId);
-    ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command);
+    ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command, 
+        object? additionalData, IDictionary<string,string>? additionalHeaders);
 
-    ValueTask<bool> TryChangeStatusAsync(IOperation operation, OperationStatus newStatus, object? additionalData);
-    ValueTask AddProgressAsync(Guid progressId, DateTimeOffset timestamp, IOperation operation, IOperationTask task, object? data);
+    ValueTask<bool> TryChangeStatusAsync(IOperation operation, OperationStatus newStatus, object? additionalData, 
+        IDictionary<string,string>? messageHeaders);
+    ValueTask AddProgressAsync(Guid progressId, DateTimeOffset timestamp, IOperation operation, IOperationTask task, 
+        object? data, IDictionary<string,string>? messageHeaders);
 }
diff --git a/src/Rebus.Operations/Rebus.Operations.Core/DefaultOperationDispatcher.cs b/src/Rebus.Operations/Rebus.Operations.Core/DefaultOperationDispatcher.cs
@@ -22,8 +22,10 @@ public DefaultOperationDispatcher(
         _operationManager = operationManager;
     }
 
-    protected override async ValueTask<(IOperation, object)> CreateOperation(object command, object? additionalData,IDictionary<string,string>? additionalHeaders)
+    protected override async ValueTask<(IOperation, object)> CreateOperation(object command, object? additionalData,
+        IDictionary<string,string>? additionalHeaders)
     {
-        return (await _operationManager.GetOrCreateAsync(Guid.NewGuid(), command), command);
+        return (await _operationManager.GetOrCreateAsync(Guid.NewGuid(), command, 
+            additionalData,additionalHeaders), command);
     }
 }
diff --git a/src/Rebus.Operations/Rebus.Operations.Core/OperationDispatcherBase.cs b/src/Rebus.Operations/Rebus.Operations.Core/OperationDispatcherBase.cs
@@ -52,7 +52,7 @@ protected OperationDispatcherBase(IBus bus, WorkflowOptions options, ILogger<Ope
 
             var(operation, taskCommand) = await CreateOperation(command, additionalData, additionalHeaders);
 
-            var commandJson = JsonSerializer.Serialize(taskCommand);
+            var commandJson = JsonSerializer.Serialize(taskCommand, _options.JsonSerializerOptions);
 
             var taskMessage = new CreateNewOperationTaskCommand(
                 taskCommand.GetType().AssemblyQualifiedName,

diff --git a/src/Rebus.Operations/Rebus.Operations.Core/OperationTaskDispatcherBase.cs b/src/Rebus.Operations/Rebus.Operations.Core/OperationTaskDispatcherBase.cs
@@ -7,7 +7,6 @@
 using System.Text.Json;
 using System.Threading.Tasks;
 using Dbosoft.Rebus.Operations.Commands;
-using Dbosoft.Rebus.Operations.Workflow;
 using Microsoft.Extensions.Logging;
 using Rebus.Bus;
 
@@ -52,7 +51,7 @@ protected OperationTaskDispatcherBase(IBus bus, WorkflowOptions options, ILogger
             throw new ArgumentNullException(nameof(command));
 
         var (task, taskCommand) = await CreateTask(operationId, initiatingTaskId, command, additionalData, additionalHeaders);
-        var commandJson = JsonSerializer.Serialize(taskCommand);
+        var commandJson = JsonSerializer.Serialize(taskCommand, _options.JsonSerializerOptions);
 
         var taskMessage = new CreateNewOperationTaskCommand(
             taskCommand.GetType().AssemblyQualifiedName,

diff --git a/src/Rebus.Operations/Rebus.Operations.Core/Workflow/OperationManagerBase.cs b/src/Rebus.Operations/Rebus.Operations.Core/Workflow/OperationManagerBase.cs
@@ -1,5 +1,6 @@
 #nullable enable
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Dbosoft.Rebus.Operations.Workflow;
@@ -9,15 +10,16 @@ public abstract class OperationManagerBase : IOperationManager
 
     public abstract ValueTask<IOperation?> GetByIdAsync(Guid operationId);
 
-    public abstract ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command);
+    public abstract ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command, 
+        object? additionalData,IDictionary<string,string>? additionalHeaders);
 
 
     public abstract ValueTask AddProgressAsync(Guid progressId, DateTimeOffset timestamp, IOperation operation,
         IOperationTask task,
-        object? data);
+        object? data, IDictionary<string,string>? messageHeaders);
 
     public abstract ValueTask<bool> TryChangeStatusAsync(IOperation operation, OperationStatus newStatus,
-        object? additionalData);
+        object? additionalData, IDictionary<string,string>? messageHeaders);
 
 
 

diff --git a/src/Rebus.Operations/Rebus.Operations.Core/Workflow/OperationTaskProgressEventHandler.cs b/src/Rebus.Operations/Rebus.Operations.Core/Workflow/OperationTaskProgressEventHandler.cs
@@ -3,6 +3,7 @@
 using JetBrains.Annotations;
 using Microsoft.Extensions.Logging;
 using Rebus.Handlers;
+using Rebus.Pipeline;
 
 namespace Dbosoft.Rebus.Operations.Workflow
 {
@@ -39,7 +40,7 @@ await _workflow.Operations.AddProgressAsync(
                     message.Timestamp,
                     operation,
                     task,
-                    message.Data);
+                    message.Data, MessageContext.Current.Headers);
             }
             else
             {

diff --git a/src/Rebus.Operations/Rebus.Operations.Core/Workflow/ProcessOperationSaga.cs b/src/Rebus.Operations/Rebus.Operations.Core/Workflow/ProcessOperationSaga.cs
@@ -8,6 +8,7 @@
 using JetBrains.Annotations;
 using Microsoft.Extensions.Logging;
 using Rebus.Handlers;
+using Rebus.Pipeline;
 using Rebus.Sagas;
 
 namespace Dbosoft.Rebus.Operations.Workflow
@@ -117,7 +118,8 @@ public async Task Handle(OperationTaskAcceptedEvent message)
             }
 
             var opOldStatus = op.Status;
-            if (await _workflow.Operations.TryChangeStatusAsync(op, OperationStatus.Running, null))
+            if (await _workflow.Operations.TryChangeStatusAsync(op, OperationStatus.Running, null, 
+                    MessageContext.Current.Headers))
             {
                 _log.LogDebug("Operation Workflow {operationId}: Status changed: {oldStatus} -> {newStatus}",
                     message.OperationId, opOldStatus, op.Status);
@@ -195,7 +197,7 @@ public async Task Handle(OperationTaskStatusEvent message)
 
 
                 if (await _workflow.Operations.TryChangeStatusAsync(op,
-                        newStatus, message.GetMessage()))
+                        newStatus, message.GetMessage(), MessageContext.Current.Headers))
                 {
                     await _workflow.Messaging.DispatchOperationStatusEventAsync(new OperationStatusEvent
                     {

diff --git a/src/Rebus.Operations/Rebus.Operations.Primitives/WorkflowOptions.cs b/src/Rebus.Operations/Rebus.Operations.Primitives/WorkflowOptions.cs
@@ -1,9 +1,17 @@
+using System.Text.Json;
+
 namespace Dbosoft.Rebus.Operations;
 
 public class WorkflowOptions
 {
+    public WorkflowOptions()
+    {
+        JsonSerializerOptions = new JsonSerializerOptions(JsonSerializerDefaults.Web);
+    }
+
     public WorkflowEventDispatchMode DispatchMode { get; set; } = WorkflowEventDispatchMode.Publish;
     public string? EventDestination { get; set; }
 
     public string? OperationsDestination { get; set; }
+    public JsonSerializerOptions JsonSerializerOptions { get; set; }
 }
diff --git a/test/Rebus.Operations.Tests/TestOperationManager.cs b/test/Rebus.Operations.Tests/TestOperationManager.cs
@@ -21,7 +21,8 @@ public static void Reset()
 
     }
 
-    public override ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command)
+    public override ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object command,
+        object? additionalData, IDictionary<string,string>? additionalHeaders)
     {
         if (Operations.ContainsKey(operationId))
             return GetByIdAsync(operationId)!;
@@ -35,7 +36,8 @@ public override ValueTask<IOperation> GetOrCreateAsync(Guid operationId, object
         return new ValueTask<IOperation>(op);
     }
 
-    public override ValueTask<bool> TryChangeStatusAsync(IOperation operation, OperationStatus newStatus, object? additionalData)
+    public override ValueTask<bool> TryChangeStatusAsync(IOperation operation, OperationStatus newStatus, 
+        object? additionalData, IDictionary<string,string>? messageHeaders)
     {
         if (!Operations.ContainsKey(operation.Id))
             return new ValueTask<bool>(false);
@@ -46,7 +48,7 @@ public override ValueTask<bool> TryChangeStatusAsync(IOperation operation, Opera
     }
 
     public override ValueTask AddProgressAsync(Guid progressId, DateTimeOffset timestamp, IOperation operation, IOperationTask task,
-        object? data)
+        object? data, IDictionary<string,string>? messageHeaders)
     {
         if(!Progress.ContainsKey(progressId))
             Progress.Add(operation.Id, new List<object>());