diff --git a/src/modules/Elsa.Common/Extensions/StringExtensions.cs b/src/modules/Elsa.Common/Extensions/StringExtensions.cs index d22cb866a..fc1b190b6 100644 --- a/src/modules/Elsa.Common/Extensions/StringExtensions.cs +++ b/src/modules/Elsa.Common/Extensions/StringExtensions.cs @@ -6,4 +6,5 @@ public static class StringExtensions public static string WithDefault(this string? value, string defaultValue) => !string.IsNullOrWhiteSpace(value) ? value : defaultValue; public static string EmptyIfNull(this string? value) => value ?? ""; public static string? NullIfEmpty(this string? value) => value == "" ? null : value; + public static string? NullIfWhiteSpace(this string? value) => string.IsNullOrWhiteSpace(value) ? null : value; } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs b/src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs index 32478f736..9f6542bbd 100644 --- a/src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs +++ b/src/modules/Elsa.Workflows.Runtime/Activities/PublishEvent.cs @@ -41,7 +41,7 @@ public class PublishEvent([CallerFilePath] string? source = null, [CallerLineNum protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { var eventName = EventName.Get(context); - var correlationId = CorrelationId.GetOrDefault(context).NullIfEmpty(); + var correlationId = CorrelationId.GetOrDefault(context).NullIfWhiteSpace(); var isLocalEvent = IsLocalEvent.GetOrDefault(context); var workflowInstanceId = isLocalEvent ? context.WorkflowExecutionContext.Id : null; var payload = Payload.GetOrDefault(context); diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/PublishEventTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/PublishEventTests.cs new file mode 100644 index 000000000..96e1f5cc2 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/PublishEventTests.cs @@ -0,0 +1,132 @@ +using System.Text.Json; +using Elsa.Common.Models; +using Elsa.Testing.Shared; +using Elsa.Testing.Shared.Services; +using Elsa.Workflows.ComponentTests.Abstractions; +using Elsa.Workflows.ComponentTests.Fixtures; +using Elsa.Workflows.ComponentTests.Scenarios.Activities.Primitives.Event.Workflows; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Messages; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Primitives.Event; + +public class PublishEventTests : AppComponentTest +{ + private readonly AsyncWorkflowRunner _workflowRunner; + private readonly IWorkflowInstanceStore _workflowInstanceStore; + private readonly IWorkflowRuntime _workflowRuntime; + private readonly WorkflowEvents _workflowEvents; + + public PublishEventTests(App app) : base(app) + { + _workflowRunner = Scope.ServiceProvider.GetRequiredService(); + _workflowInstanceStore = Scope.ServiceProvider.GetRequiredService(); + _workflowRuntime = Scope.ServiceProvider.GetRequiredService(); + _workflowEvents = Scope.ServiceProvider.GetRequiredService(); + } + + [Fact] + public async Task PublishEvent_LocalEvent_CompletesWorkflow() + { + // Act + var result = await _workflowRunner.RunAndAwaitWorkflowCompletionAsync(WorkflowDefinitionHandle.ByDefinitionId(PublishAndConsumeEventWorkflow.DefinitionId, VersionOptions.Published)); + + // Assert + Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowExecutionContext.SubStatus); + } + + [Fact] + public async Task PublishEvent_GlobalEvent_TriggersConsumerWorkflow() + { + // Arrange + var correlationId = Guid.NewGuid().ToString(); + + // Act - Publish global event + await RunWorkflowAsync(PublishGlobalEventWorkflow.DefinitionId, correlationId); + + // Assert - Consumer workflow was triggered and completed + var consumerInstance = await GetSingleWorkflowInstanceAsync(ConsumerWorkflow.DefinitionId, correlationId); + Assert.Equal(WorkflowStatus.Finished, consumerInstance.Status); + Assert.Equal(WorkflowSubStatus.Finished, consumerInstance.SubStatus); + } + + [Fact] + public async Task PublishEvent_WithPayload_TransmitsPayloadToConsumer() + { + // Arrange + var correlationId = Guid.NewGuid().ToString(); + + // Act - Publish global event with payload + await RunWorkflowAsync(PublishGlobalEventWorkflow.DefinitionId, correlationId); + + // Assert - Consumer workflow received the payload + var consumerInstance = await GetSingleWorkflowInstanceAsync(ConsumerWorkflow.DefinitionId, correlationId); + Assert.Equal(WorkflowStatus.Finished, consumerInstance.Status); + Assert.Equal(WorkflowSubStatus.Finished, consumerInstance.SubStatus); + + // Verify the payload was captured in the output + Assert.True(consumerInstance.WorkflowState.Output.TryGetValue("ReceivedPayload", out var receivedPayload), "Consumer workflow should have ReceivedPayload output"); + Assert.NotNull(receivedPayload); + + // Verify the payload structure and content + var payloadJson = JsonSerializer.Serialize(receivedPayload); + Assert.Contains("\"Status\"", payloadJson); + Assert.Contains("\"Shipped\"", payloadJson); + } + + private async Task RunWorkflowAsync(string definitionId, string? correlationId = null) + { + var workflowClient = await _workflowRuntime.CreateClientAsync(); + await workflowClient.CreateInstanceAsync(new() + { + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, VersionOptions.Published), + CorrelationId = correlationId + }); + await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty); + } + + private async Task GetSingleWorkflowInstanceAsync(string definitionId, string correlationId, int timeoutMs = 5000) + { + var tcs = new TaskCompletionSource(); + var cts = new CancellationTokenSource(timeoutMs); + + // Register cancellation to fail the task on timeout + cts.Token.Register(() => tcs.TrySetException(new TimeoutException($"Workflow instance with DefinitionId '{definitionId}' and CorrelationId '{correlationId}' was not saved within {timeoutMs}ms"))); + + // Subscribe to the WorkflowInstanceSaved event + void OnWorkflowInstanceSaved(object? sender, WorkflowInstanceSavedEventArgs args) + { + if (args.WorkflowInstance.DefinitionId == definitionId && args.WorkflowInstance.CorrelationId == correlationId) + { + tcs.TrySetResult(args.WorkflowInstance); + } + } + + _workflowEvents.WorkflowInstanceSaved += OnWorkflowInstanceSaved; + + try + { + // Check if the instance already exists in the database + var existingInstances = (await _workflowInstanceStore.FindManyAsync(new() + { + DefinitionId = definitionId, + CorrelationId = correlationId + }, cts.Token)).ToList(); + + if (existingInstances.Any()) + return Assert.Single(existingInstances); + + // Wait for the event to be raised + return await tcs.Task; + } + finally + { + _workflowEvents.WorkflowInstanceSaved -= OnWorkflowInstanceSaved; + cts.Dispose(); + } + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/ConsumerWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/ConsumerWorkflow.cs new file mode 100644 index 000000000..547460c07 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/ConsumerWorkflow.cs @@ -0,0 +1,42 @@ +using Elsa.Workflows.Activities; +using Elsa.Workflows.Management.Activities.SetOutput; +using Elsa.Workflows.Memory; + +namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Primitives.Event.Workflows; + +/// +/// A workflow that listens for global events as a trigger and captures the payload. +/// +public class ConsumerWorkflow : WorkflowBase +{ + public static readonly string DefinitionId = Guid.NewGuid().ToString(); + + protected override void Build(IWorkflowBuilder workflow) + { + var eventPayload = new Variable(); + + workflow.WithDefinitionId(DefinitionId); + workflow.WithVariable(eventPayload); + workflow.WithOutput("ReceivedPayload"); // Declare the output + + var eventActivity = new Runtime.Activities.Event("GlobalOrderEvent") + { + CanStartWorkflow = true, + Result = new(eventPayload) + }; + + workflow.Root = new Sequence + { + Activities = + { + eventActivity, + new SetOutput + { + OutputName = new("ReceivedPayload"), + OutputValue = new(eventPayload) + }, + new End() + } + }; + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishAndConsumeEventWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishAndConsumeEventWorkflow.cs new file mode 100644 index 000000000..b51d53764 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishAndConsumeEventWorkflow.cs @@ -0,0 +1,34 @@ +using Elsa.Workflows.Activities; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Primitives.Event.Workflows; + +/// +/// A workflow that publishes an event to itself (local event). +/// +public class PublishAndConsumeEventWorkflow : WorkflowBase +{ + public static readonly string DefinitionId = Guid.NewGuid().ToString(); + + protected override void Build(IWorkflowBuilder workflow) + { + workflow.WithDefinitionId(DefinitionId); + workflow.Root = new Sequence + { + Activities = + { + new Start(), + // Publish a local event + new PublishEvent + { + EventName = new("LocalOrderEvent"), + IsLocalEvent = new(true), + Payload = new(new { OrderId = 123 }) + }, + // Wait for the local event + new Elsa.Workflows.Runtime.Activities.Event("LocalOrderEvent"), + new End() + } + }; + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishGlobalEventWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishGlobalEventWorkflow.cs new file mode 100644 index 000000000..4fbbb7824 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Primitives/Event/Workflows/PublishGlobalEventWorkflow.cs @@ -0,0 +1,33 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Runtime.Activities; + +namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Primitives.Event.Workflows; + +/// +/// A workflow that publishes a global event with the workflow's correlation ID. +/// +public class PublishGlobalEventWorkflow : WorkflowBase +{ + public static readonly string DefinitionId = Guid.NewGuid().ToString(); + + protected override void Build(IWorkflowBuilder workflow) + { + workflow.WithDefinitionId(DefinitionId); + workflow.Root = new Sequence + { + Activities = + { + new Start(), + new PublishEvent + { + EventName = new("GlobalOrderEvent"), + CorrelationId = new(context => context.GetWorkflowExecutionContext().CorrelationId), + IsLocalEvent = new(false), + Payload = new(new { Status = "Shipped" }) + }, + new End() + } + }; + } +} diff --git a/test/unit/Elsa.Activities.UnitTests/Event/EventBaseTests.cs b/test/unit/Elsa.Activities.UnitTests/Primitives/EventBaseTests.cs similarity index 97% rename from test/unit/Elsa.Activities.UnitTests/Event/EventBaseTests.cs rename to test/unit/Elsa.Activities.UnitTests/Primitives/EventBaseTests.cs index 9d1248519..c952718b9 100644 --- a/test/unit/Elsa.Activities.UnitTests/Event/EventBaseTests.cs +++ b/test/unit/Elsa.Activities.UnitTests/Primitives/EventBaseTests.cs @@ -6,7 +6,7 @@ using Elsa.Workflows.Runtime; using Elsa.Workflows.Runtime.Activities; using Elsa.Workflows.Runtime.Stimuli; -namespace Elsa.Activities.UnitTests.Event; +namespace Elsa.Activities.UnitTests.Primitives; public class EventBaseTests { @@ -64,7 +64,7 @@ public class EventBaseTests // Act var context = await ExecuteAsync(activity); - context.WorkflowExecutionContext.Input[Elsa.Workflows.Runtime.Activities.Event.EventInputWorkflowInputKey] = expectedInput; + context.WorkflowExecutionContext.Input[Event.EventInputWorkflowInputKey] = expectedInput; await activity.InvokeCallbackAsync(context); // Assert diff --git a/test/unit/Elsa.Activities.UnitTests/Primitives/PublishEventTests.cs b/test/unit/Elsa.Activities.UnitTests/Primitives/PublishEventTests.cs new file mode 100644 index 000000000..255fd34a6 --- /dev/null +++ b/test/unit/Elsa.Activities.UnitTests/Primitives/PublishEventTests.cs @@ -0,0 +1,148 @@ +using Elsa.Testing.Shared; +using Elsa.Workflows; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Activities; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; + +namespace Elsa.Activities.UnitTests.Primitives; + +public class PublishEventTests +{ + [Theory] + [InlineData("OrderCreated", null, null, false)] + [InlineData("OrderCreated", "correlation-123", "correlation-123", false)] + [InlineData("OrderEvent", "", null, true)] + [InlineData("OrderEvent", " ", null, true)] + public async Task ExecuteAsync_PublishesEvent_WithParameters(string eventName, string? correlationId, string? expectedCorrelationId, bool expectNullCorrelation) + { + // Arrange + var publisher = Substitute.For(); + + // Act + await ExecuteAsync(CreateActivity(eventName, correlationId), publisher); + + // Assert + await AssertPublishedAsync( + publisher, + eventName, + correlationId: expectedCorrelationId, + expectNullCorrelationId: expectNullCorrelation); + } + + [Theory] + [InlineData(true, false)] + [InlineData(false, true)] + public async Task ExecuteAsync_LocalEvent_PassesCorrectWorkflowInstanceId(bool isLocalEvent, bool expectNull) + { + // Arrange + const string eventName = "TestEvent"; + var publisher = Substitute.For(); + + // Act + var context = await ExecuteAsync(CreateActivity(eventName, isLocalEvent: isLocalEvent), publisher); + + // Assert + await AssertPublishedAsync( + publisher, + eventName, + workflowInstanceId: expectNull ? null : context.WorkflowExecutionContext.Id, + expectNullWorkflowInstanceId: expectNull); + } + + [Fact] + public async Task ExecuteAsync_PublishesEvent_WithPayload() + { + // Arrange + const string eventName = "OrderCreated"; + var payload = new { OrderId = 123, Amount = 99.99m }; + var publisher = Substitute.For(); + + // Act + await ExecuteAsync(CreateActivity(eventName, payload: payload), publisher); + + // Assert + await AssertPublishedAsync(publisher, eventName, payload: payload); + } + + [Fact] + public async Task ExecuteAsync_CompletesActivity() + { + // Arrange + var publisher = Substitute.For(); + + // Act + var context = await ExecuteAsync(CreateActivity("TestEvent"), publisher); + + // Assert + Assert.Equal(ActivityStatus.Completed, context.Status); + } + + [Fact] + public async Task ExecuteAsync_WithAllParameters_PassesAllValuesToPublisher() + { + // Arrange + const string eventName = "CompleteOrderEvent"; + const string correlationId = "correlation-456"; + var payload = new { Status = "Shipped" }; + var publisher = Substitute.For(); + + // Act + var context = await ExecuteAsync(CreateActivity(eventName, correlationId, isLocalEvent: true, payload: payload), publisher); + + // Assert + await publisher.Received(1).PublishAsync( + eventName, + correlationId, + context.WorkflowExecutionContext.Id, + null, + payload, + true, + Arg.Any()); + } + + private static PublishEvent CreateActivity( + string eventName, + string? correlationId = null, + bool? isLocalEvent = null, + object? payload = null) => + new() + { + EventName = new(eventName), + CorrelationId = correlationId != null ? new(correlationId) : null!, + IsLocalEvent = isLocalEvent.HasValue ? new(isLocalEvent.Value) : null!, + Payload = payload != null ? new(payload) : null! + }; + + private static async Task ExecuteAsync(PublishEvent activity, IEventPublisher publisher) => + await new ActivityTestFixture(activity) + .ConfigureServices(services => services.AddSingleton(publisher)) + .ExecuteAsync(); + + private static async Task AssertPublishedAsync( + IEventPublisher publisher, + string eventName, + string? correlationId = null, + string? workflowInstanceId = null, + object? payload = null, + bool expectNullCorrelationId = false, + bool expectNullWorkflowInstanceId = false) + { + var correlationIdArg = expectNullCorrelationId + ? Arg.Is(x => x == null) + : correlationId ?? Arg.Any(); + + var workflowInstanceIdArg = expectNullWorkflowInstanceId + ? Arg.Is(x => x == null) + : workflowInstanceId ?? Arg.Any(); + + await publisher.Received(1).PublishAsync( + eventName, + correlationIdArg, + workflowInstanceIdArg, + Arg.Any(), + payload ?? Arg.Any(), + Arg.Any(), + Arg.Any()); + } +}