diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 253d321bd..0a72e4cb2 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -83,6 +83,7 @@ jobs: 8.0.x 9.0.x 10.0.x + - name: Restore dependencies run: dotnet restore - name: Build run: dotnet build --no-restore @@ -294,3 +295,33 @@ jobs: with: name: oracle-test-results path: test-results/ + Azure-Tests: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - name: Setup .NET + uses: actions/setup-dotnet@v1 + with: + dotnet-version: | + 8.0.x + 9.0.x + - name: Restore dependencies + run: dotnet restore + - name: Build + run: dotnet build --no-restore + - name: Azure Tests + run: dotnet test test/WorkflowCore.Tests.Azure --no-build --verbosity detailed --logger "trx;LogFileName=AzureTests.trx" --logger "console;verbosity=detailed" --results-directory ./test-results -p:ParallelizeTestCollections=false + - name: Publish Test Results + uses: dorny/test-reporter@v1 + if: success() || failure() + with: + name: Azure Test Results + path: test-results/*.trx + reporter: dotnet-trx + fail-on-error: false + - name: Upload Test Results + uses: actions/upload-artifact@v4 + if: always() + with: + name: azure-test-results + path: test-results/ diff --git a/WorkflowCore.sln b/WorkflowCore.sln index 685218c34..597708202 100644 --- a/WorkflowCore.sln +++ b/WorkflowCore.sln @@ -158,6 +158,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Persistence.Or EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.Oracle", "test\WorkflowCore.Tests.Oracle\WorkflowCore.Tests.Oracle.csproj", "{A2837F1C-3740-4375-9069-81AE32C867CA}" EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.Azure", "test\WorkflowCore.Tests.Azure\WorkflowCore.Tests.Azure.csproj", "{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}" +EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.AI.AzureFoundry", "src\extensions\WorkflowCore.AI.AzureFoundry\WorkflowCore.AI.AzureFoundry.csproj", "{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.AI.AzureFoundry.Tests", "test\WorkflowCore.AI.AzureFoundry.Tests\WorkflowCore.AI.AzureFoundry.Tests.csproj", "{AAB9DDFF-0A0A-43BE-BF00-2BA13ED526C8}" @@ -846,6 +848,10 @@ Global {A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x64.Build.0 = Release|Any CPU {A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x86.ActiveCfg = Release|Any CPU {A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x86.Build.0 = Release|Any CPU + {D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Debug|Any CPU.Build.0 = Debug|Any CPU + {D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Release|Any CPU.ActiveCfg = Release|Any CPU + {D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Release|Any CPU.Build.0 = Release|Any CPU {A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|Any CPU.Build.0 = Debug|Any CPU {A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|x64.ActiveCfg = Debug|Any CPU @@ -946,6 +952,7 @@ Global {AF205715-C8B7-42EF-BF14-AFC9E7F27242} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2} {635629BC-9D5C-40C6-BBD0-060550ECE290} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2} {A2837F1C-3740-4375-9069-81AE32C867CA} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB} + {D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB} {A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1} = {6803696C-B19A-4B27-9193-082A02B6F205} {AAB9DDFF-0A0A-43BE-BF00-2BA13ED526C8} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB} {D75C8112-6A4D-4A13-BB79-2D23DF66E4CB} = {5080DB09-CBE8-4C45-9957-C3BB7651755E} diff --git a/docs/persistence.md b/docs/persistence.md index 00799090a..fe8e1a438 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -10,5 +10,56 @@ There are several persistence providers available as separate Nuget packages. * [Sqlite](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Sqlite) * [Amazon DynamoDB](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.AWS) * [Cosmos DB](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Azure) +* [Azure Table Storage](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Azure) * [Redis](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Redis) -* [Oracle](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Oracle) \ No newline at end of file +* [Oracle](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Oracle) + +## Implementing a custom persistence provider + +To implement a custom persistence provider, create a class that implements `IPersistenceProvider` interface: + +```csharp +public interface IPersistenceProvider : IWorkflowRepository, ISubscriptionRepository, IEventRepository, IScheduledCommandRepository +{ + Task PersistErrors(IEnumerable errors, CancellationToken cancellationToken = default); + void EnsureStoreExists(); +} +``` + +The `IPersistenceProvider` interface combines four repository interfaces: + +### IWorkflowRepository +Handles workflow instance storage and retrieval: +- `CreateNewWorkflow` - Create and store a new workflow instance +- `PersistWorkflow` - Update an existing workflow instance +- `GetWorkflowInstance` - Retrieve a specific workflow instance +- `GetRunnableInstances` - Get workflow instances ready for execution + +### IEventRepository +Manages workflow events: +- `CreateEvent` - Store a new event +- `GetEvent` - Retrieve a specific event +- `GetRunnableEvents` - Get events ready for processing +- `MarkEventProcessed/Unprocessed` - Update event status + +### ISubscriptionRepository +Handles event subscriptions: +- `CreateEventSubscription` - Create new event subscription +- `GetSubscriptions` - Query subscriptions for events +- `TerminateSubscription` - Remove a subscription +- `SetSubscriptionToken/ClearSubscriptionToken` - Manage subscription locking + +### IScheduledCommandRepository +For future command scheduling (optional): +- `ScheduleCommand` - Schedule a command for future execution +- `ProcessCommands` - Execute scheduled commands +- `SupportsScheduledCommands` - Indicates if provider supports this feature + +Once implemented, register your provider: + +```csharp +services.AddWorkflow(options => +{ + options.UsePersistence(sp => new MyCustomPersistenceProvider()); +}); +``` \ No newline at end of file diff --git a/src/WorkflowCore/Services/LifeCycleEventPublisher.cs b/src/WorkflowCore/Services/LifeCycleEventPublisher.cs index e73c6bf4f..3634d3cbb 100644 --- a/src/WorkflowCore/Services/LifeCycleEventPublisher.cs +++ b/src/WorkflowCore/Services/LifeCycleEventPublisher.cs @@ -26,10 +26,16 @@ public LifeCycleEventPublisher(ILifeCycleEventHub eventHub, WorkflowOptions work public void PublishNotification(LifeCycleEvent evt) { - if (_outbox.IsAddingCompleted || !_workflowOptions.EnableLifeCycleEventsPublisher) + if (!_workflowOptions.EnableLifeCycleEventsPublisher) return; - _outbox.Add(evt); + try + { + _outbox.TryAdd(evt); + } + catch (ObjectDisposedException) + { + } } public void Start() @@ -44,8 +50,7 @@ public void Start() _outbox = new BlockingCollection(); } - _dispatchTask = new Task(Execute); - _dispatchTask.Start(); + _dispatchTask = Task.Run(Execute); } public void Stop() @@ -57,22 +62,36 @@ public void Stop() public void Dispose() { + if (_dispatchTask != null) + { + if (!_outbox.IsAddingCompleted) + _outbox.CompleteAdding(); + if (!_dispatchTask.Wait(TimeSpan.FromSeconds(30))) + _logger.LogWarning("Lifecycle event publisher did not stop within timeout"); + _dispatchTask = null; + } _outbox.Dispose(); } - private async void Execute() + private async Task Execute() { - foreach (var evt in _outbox.GetConsumingEnumerable()) + try { - try - { - await _eventHub.PublishNotification(evt); - } - catch (Exception ex) + foreach (var evt in _outbox.GetConsumingEnumerable()) { - _logger.LogError(default(EventId), ex, ex.Message); + try + { + await _eventHub.PublishNotification(evt); + } + catch (Exception ex) + { + _logger.LogError(default(EventId), ex, ex.Message); + } } } + catch (ObjectDisposedException) + { + } } } } \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/Models/EventTableEntity.cs b/src/providers/WorkflowCore.Providers.Azure/Models/EventTableEntity.cs new file mode 100644 index 000000000..cee71495d --- /dev/null +++ b/src/providers/WorkflowCore.Providers.Azure/Models/EventTableEntity.cs @@ -0,0 +1,54 @@ +using System; +using Azure; +using Azure.Data.Tables; +using Newtonsoft.Json; +using WorkflowCore.Models; + +namespace WorkflowCore.Providers.Azure.Models +{ + public class EventTableEntity : ITableEntity + { + public string PartitionKey { get; set; } + public string RowKey { get; set; } + public DateTimeOffset? Timestamp { get; set; } + public ETag ETag { get; set; } + + public string EventName { get; set; } + public string EventKey { get; set; } + public string EventData { get; set; } + public DateTime EventTime { get; set; } + public bool IsProcessed { get; set; } + + private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }; + + public static EventTableEntity FromInstance(Event instance) + { + return new EventTableEntity + { + PartitionKey = "event", + RowKey = instance.Id, + EventName = instance.EventName, + EventKey = instance.EventKey, + EventTime = EnsureUtc(instance.EventTime), + IsProcessed = instance.IsProcessed, + EventData = JsonConvert.SerializeObject(instance.EventData, SerializerSettings), + }; + } + + private static DateTime EnsureUtc(DateTime dt) => + dt.Kind == DateTimeKind.Utc ? dt : DateTime.SpecifyKind(dt, DateTimeKind.Utc); + + public static Event ToInstance(EventTableEntity entity) + { + return new Event + { + Id = entity.RowKey, + EventName = entity.EventName, + EventKey = entity.EventKey, + EventTime = entity.EventTime, + IsProcessed = entity.IsProcessed, + EventData = JsonConvert.DeserializeObject(entity.EventData, SerializerSettings), + }; + } + } +} \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/Models/ScheduledCommandTableEntity.cs b/src/providers/WorkflowCore.Providers.Azure/Models/ScheduledCommandTableEntity.cs new file mode 100644 index 000000000..daea36ac2 --- /dev/null +++ b/src/providers/WorkflowCore.Providers.Azure/Models/ScheduledCommandTableEntity.cs @@ -0,0 +1,44 @@ +using System; +using Azure; +using Azure.Data.Tables; +using Newtonsoft.Json; +using WorkflowCore.Models; + +namespace WorkflowCore.Providers.Azure.Models +{ + public class ScheduledCommandTableEntity : ITableEntity + { + public string PartitionKey { get; set; } + public string RowKey { get; set; } + public DateTimeOffset? Timestamp { get; set; } + public ETag ETag { get; set; } + + public string CommandName { get; set; } + public string Data { get; set; } + public long ExecuteTime { get; set; } + + private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }; + + public static ScheduledCommandTableEntity FromInstance(ScheduledCommand instance, string rowKey = null) + { + return new ScheduledCommandTableEntity + { + PartitionKey = "command", + RowKey = rowKey ?? Guid.NewGuid().ToString(), + CommandName = instance.CommandName, + Data = instance.Data, + ExecuteTime = instance.ExecuteTime, + }; + } + + public static ScheduledCommand ToInstance(ScheduledCommandTableEntity entity) + { + return new ScheduledCommand + { + CommandName = entity.CommandName, + Data = entity.Data, + ExecuteTime = entity.ExecuteTime, + }; + } + } +} \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/Models/SubscriptionTableEntity.cs b/src/providers/WorkflowCore.Providers.Azure/Models/SubscriptionTableEntity.cs new file mode 100644 index 000000000..811969123 --- /dev/null +++ b/src/providers/WorkflowCore.Providers.Azure/Models/SubscriptionTableEntity.cs @@ -0,0 +1,69 @@ +using System; +using Azure; +using Azure.Data.Tables; +using Newtonsoft.Json; +using WorkflowCore.Models; + +namespace WorkflowCore.Providers.Azure.Models +{ + public class SubscriptionTableEntity : ITableEntity + { + public string PartitionKey { get; set; } + public string RowKey { get; set; } + public DateTimeOffset? Timestamp { get; set; } + public ETag ETag { get; set; } + + public string WorkflowId { get; set; } + public int StepId { get; set; } + public string ExecutionPointerId { get; set; } + public string EventName { get; set; } + public string EventKey { get; set; } + public DateTime SubscribeAsOf { get; set; } + public string SubscriptionData { get; set; } + public string ExternalToken { get; set; } + public string ExternalWorkerId { get; set; } + public DateTime? ExternalTokenExpiry { get; set; } + + private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }; + + public static SubscriptionTableEntity FromInstance(EventSubscription instance) + { + return new SubscriptionTableEntity + { + PartitionKey = "subscription", + RowKey = instance.Id, + WorkflowId = instance.WorkflowId, + StepId = instance.StepId, + ExecutionPointerId = instance.ExecutionPointerId, + EventName = instance.EventName, + EventKey = instance.EventKey, + SubscribeAsOf = EnsureUtc(instance.SubscribeAsOf), + ExternalToken = instance.ExternalToken, + ExternalWorkerId = instance.ExternalWorkerId, + ExternalTokenExpiry = instance.ExternalTokenExpiry.HasValue ? EnsureUtc(instance.ExternalTokenExpiry.Value) : (DateTime?)null, + SubscriptionData = JsonConvert.SerializeObject(instance.SubscriptionData, SerializerSettings), + }; + } + + private static DateTime EnsureUtc(DateTime dt) => + dt.Kind == DateTimeKind.Utc ? dt : DateTime.SpecifyKind(dt, DateTimeKind.Utc); + + public static EventSubscription ToInstance(SubscriptionTableEntity entity) + { + return new EventSubscription + { + Id = entity.RowKey, + WorkflowId = entity.WorkflowId, + StepId = entity.StepId, + ExecutionPointerId = entity.ExecutionPointerId, + EventName = entity.EventName, + EventKey = entity.EventKey, + SubscribeAsOf = entity.SubscribeAsOf, + ExternalToken = entity.ExternalToken, + ExternalWorkerId = entity.ExternalWorkerId, + ExternalTokenExpiry = entity.ExternalTokenExpiry, + SubscriptionData = JsonConvert.DeserializeObject(entity.SubscriptionData, SerializerSettings), + }; + } + } +} \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/Models/WorkflowTableEntity.cs b/src/providers/WorkflowCore.Providers.Azure/Models/WorkflowTableEntity.cs new file mode 100644 index 000000000..009375c84 --- /dev/null +++ b/src/providers/WorkflowCore.Providers.Azure/Models/WorkflowTableEntity.cs @@ -0,0 +1,69 @@ +using System; +using Azure; +using Azure.Data.Tables; +using Newtonsoft.Json; +using WorkflowCore.Models; + +namespace WorkflowCore.Providers.Azure.Models +{ + public class WorkflowTableEntity : ITableEntity + { + public string PartitionKey { get; set; } + public string RowKey { get; set; } + public DateTimeOffset? Timestamp { get; set; } + public ETag ETag { get; set; } + + public string WorkflowDefinitionId { get; set; } + public int Version { get; set; } + public string Description { get; set; } + public string Reference { get; set; } + public string ExecutionPointers { get; set; } + public long? NextExecution { get; set; } + public int Status { get; set; } + public string Data { get; set; } + public DateTime CreateTime { get; set; } + public DateTime? CompleteTime { get; set; } + + private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }; + + public static WorkflowTableEntity FromInstance(WorkflowInstance instance) + { + return new WorkflowTableEntity + { + PartitionKey = "workflow", + RowKey = instance.Id, + WorkflowDefinitionId = instance.WorkflowDefinitionId, + Version = instance.Version, + Description = instance.Description, + Reference = instance.Reference, + NextExecution = instance.NextExecution, + Status = (int)instance.Status, + CreateTime = EnsureUtc(instance.CreateTime), + CompleteTime = instance.CompleteTime.HasValue ? EnsureUtc(instance.CompleteTime.Value) : (DateTime?)null, + Data = JsonConvert.SerializeObject(instance.Data, SerializerSettings), + ExecutionPointers = JsonConvert.SerializeObject(instance.ExecutionPointers, SerializerSettings), + }; + } + + private static DateTime EnsureUtc(DateTime dt) => + dt.Kind == DateTimeKind.Utc ? dt : DateTime.SpecifyKind(dt, DateTimeKind.Utc); + + public static WorkflowInstance ToInstance(WorkflowTableEntity entity) + { + return new WorkflowInstance + { + Id = entity.RowKey, + WorkflowDefinitionId = entity.WorkflowDefinitionId, + Version = entity.Version, + Description = entity.Description, + Reference = entity.Reference, + NextExecution = entity.NextExecution, + Status = (WorkflowStatus)entity.Status, + CreateTime = entity.CreateTime, + CompleteTime = entity.CompleteTime, + Data = JsonConvert.DeserializeObject(entity.Data, SerializerSettings), + ExecutionPointers = JsonConvert.DeserializeObject(entity.ExecutionPointers, SerializerSettings), + }; + } + } +} \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/README.md b/src/providers/WorkflowCore.Providers.Azure/README.md index 10c17bfbe..e7be5c58f 100644 --- a/src/providers/WorkflowCore.Providers.Azure/README.md +++ b/src/providers/WorkflowCore.Providers.Azure/README.md @@ -4,6 +4,7 @@ * Provides Queueing support on [Workflow Core](../../README.md) using Azure Storage queues. * Provides event hub support on [Workflow Core](../../README.md) backed by Azure Service Bus. * Provides persistence on [Workflow Core](../../README.md) backed by Azure Cosmos DB. +* Provides persistence on [Workflow Core](../../README.md) backed by Azure Table Storage. This makes it possible to have a cluster of nodes processing your workflows. @@ -25,6 +26,8 @@ dotnet add package WorkflowCore.Providers.Azure Use the `IServiceCollection` extension methods when building your service provider * .UseAzureSynchronization * .UseAzureServiceBusEventHub +* .UseCosmosDbPersistence +* .UseAzureTableStoragePersistence ```C# services.AddWorkflow(options => @@ -33,4 +36,35 @@ services.AddWorkflow(options => options.UseAzureServiceBusEventHub("service bus connection string", "topic name", "subscription name"); options.UseCosmosDbPersistence("connection string"); }); +``` + +### Azure Table Storage Persistence + +For cost-effective workflow persistence using Azure Table Storage: + +```C# +services.AddWorkflow(options => +{ + options.UseAzureTableStoragePersistence("azure storage connection string"); +}); +``` + +You can also specify a custom table name prefix: + +```C# +services.AddWorkflow(options => +{ + options.UseAzureTableStoragePersistence("azure storage connection string", "MyWorkflows"); +}); +``` + +Or use with managed identity: + +```C# +services.AddWorkflow(options => +{ + var tableServiceUri = new Uri("https://mystorageaccount.table.core.windows.net"); + var credential = new DefaultAzureCredential(); + options.UseAzureTableStoragePersistence(tableServiceUri, credential); +}); ``` \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/ServiceCollectionExtensions.cs b/src/providers/WorkflowCore.Providers.Azure/ServiceCollectionExtensions.cs index e08d0445f..0178ab39b 100644 --- a/src/providers/WorkflowCore.Providers.Azure/ServiceCollectionExtensions.cs +++ b/src/providers/WorkflowCore.Providers.Azure/ServiceCollectionExtensions.cs @@ -1,5 +1,6 @@ using System; using Azure.Core; +using Azure.Data.Tables; using Microsoft.Azure.Cosmos; using Microsoft.Extensions.Logging; using WorkflowCore.Interface; @@ -106,5 +107,36 @@ public static WorkflowOptions UseCosmosDbPersistence( options.UsePersistence(sp => new CosmosDbPersistenceProvider(sp.GetService(), databaseId, sp.GetService(), cosmosDbStorageOptions)); return options; } + + public static WorkflowOptions UseAzureTableStoragePersistence( + this WorkflowOptions options, + string connectionString, + string tableNamePrefix = "WorkflowCore") + { + options.Services.AddSingleton(sp => new TableServiceClient(connectionString)); + options.UsePersistence(sp => new AzureTableStoragePersistenceProvider(sp.GetService(), tableNamePrefix, sp.GetService())); + return options; + } + + public static WorkflowOptions UseAzureTableStoragePersistence( + this WorkflowOptions options, + TableServiceClient tableServiceClient, + string tableNamePrefix = "WorkflowCore") + { + options.Services.AddSingleton(tableServiceClient); + options.UsePersistence(sp => new AzureTableStoragePersistenceProvider(sp.GetService(), tableNamePrefix, sp.GetService())); + return options; + } + + public static WorkflowOptions UseAzureTableStoragePersistence( + this WorkflowOptions options, + Uri serviceUri, + TokenCredential tokenCredential, + string tableNamePrefix = "WorkflowCore") + { + options.Services.AddSingleton(sp => new TableServiceClient(serviceUri, tokenCredential)); + options.UsePersistence(sp => new AzureTableStoragePersistenceProvider(sp.GetService(), tableNamePrefix, sp.GetService())); + return options; + } } } diff --git a/src/providers/WorkflowCore.Providers.Azure/Services/AzureTableStoragePersistenceProvider.cs b/src/providers/WorkflowCore.Providers.Azure/Services/AzureTableStoragePersistenceProvider.cs new file mode 100644 index 000000000..33736812a --- /dev/null +++ b/src/providers/WorkflowCore.Providers.Azure/Services/AzureTableStoragePersistenceProvider.cs @@ -0,0 +1,467 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using Microsoft.Extensions.Logging; +using WorkflowCore.Interface; +using WorkflowCore.Models; +using WorkflowCore.Providers.Azure.Models; + +namespace WorkflowCore.Providers.Azure.Services +{ + public class AzureTableStoragePersistenceProvider : IPersistenceProvider + { + private readonly TableServiceClient _tableServiceClient; + private readonly ILogger _logger; + private readonly string _workflowTableName; + private readonly string _eventTableName; + private readonly string _subscriptionTableName; + private readonly string _commandTableName; + private readonly string _errorTableName; + + private readonly Lazy _workflowTable; + private readonly Lazy _eventTable; + private readonly Lazy _subscriptionTable; + private readonly Lazy _commandTable; + private readonly Lazy _errorTable; + + public AzureTableStoragePersistenceProvider( + TableServiceClient tableServiceClient, + string tableNamePrefix = "WorkflowCore", + ILoggerFactory loggerFactory = null) + { + _tableServiceClient = tableServiceClient; + _logger = loggerFactory?.CreateLogger(); + _workflowTableName = $"{tableNamePrefix}Workflows"; + _eventTableName = $"{tableNamePrefix}Events"; + _subscriptionTableName = $"{tableNamePrefix}Subscriptions"; + _commandTableName = $"{tableNamePrefix}Commands"; + _errorTableName = $"{tableNamePrefix}Errors"; + + _workflowTable = new Lazy(() => _tableServiceClient.GetTableClient(_workflowTableName)); + _eventTable = new Lazy(() => _tableServiceClient.GetTableClient(_eventTableName)); + _subscriptionTable = new Lazy(() => _tableServiceClient.GetTableClient(_subscriptionTableName)); + _commandTable = new Lazy(() => _tableServiceClient.GetTableClient(_commandTableName)); + _errorTable = new Lazy(() => _tableServiceClient.GetTableClient(_errorTableName)); + } + + public bool SupportsScheduledCommands => true; + + /// Escapes single quotes in OData string literals by doubling them. + private static string EscapeOData(string value) => value?.Replace("'", "''") ?? string.Empty; + + /// + /// Safely converts a DateTime to UTC. If Kind is Unspecified, assumes UTC + /// rather than local time (avoids failures with DateTime.MinValue). + /// + private static DateTime SafeToUtc(DateTime dt) => + dt.Kind == DateTimeKind.Utc ? dt : + dt.Kind == DateTimeKind.Unspecified ? DateTime.SpecifyKind(dt, DateTimeKind.Utc) : + dt.ToUniversalTime(); + + public async Task CreateNewWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default) + { + workflow.Id = Guid.NewGuid().ToString(); + var entity = WorkflowTableEntity.FromInstance(workflow); + await _workflowTable.Value.AddEntityAsync(entity, cancellationToken); + return workflow.Id; + } + + public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default) + { + var entity = WorkflowTableEntity.FromInstance(workflow); + await _workflowTable.Value.UpsertEntityAsync(entity, TableUpdateMode.Replace, cancellationToken); + } + + public async Task PersistWorkflow(WorkflowInstance workflow, List subscriptions, CancellationToken cancellationToken = default) + { + await PersistWorkflow(workflow, cancellationToken); + + foreach (var subscription in subscriptions) + { + await CreateEventSubscription(subscription, cancellationToken); + } + } + + public async Task> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default) + { + var utcTicks = SafeToUtc(asAt).Ticks; + var query = _workflowTable.Value.QueryAsync( + filter: $"PartitionKey eq 'workflow' and Status eq {(int)WorkflowStatus.Runnable} and NextExecution le {utcTicks}", + cancellationToken: cancellationToken); + + var result = new List(); + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(cancellationToken); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + result.Add(entity.RowKey); + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + return result; + } + + public async Task GetWorkflowInstance(string id, CancellationToken cancellationToken = default) + { + try + { + var response = await _workflowTable.Value.GetEntityAsync("workflow", id, cancellationToken: cancellationToken); + return WorkflowTableEntity.ToInstance(response.Value); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return null; + } + } + + public async Task> GetWorkflowInstances(IEnumerable ids, CancellationToken cancellationToken = default) + { + if (ids == null) + return Enumerable.Empty(); + + var result = new List(); + foreach (var id in ids) + { + var instance = await GetWorkflowInstance(id, cancellationToken); + if (instance != null) + result.Add(instance); + } + return result; + } + + [Obsolete] + public async Task> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take) + { + var filter = "PartitionKey eq 'workflow'"; + + if (status.HasValue) + filter += $" and Status eq {(int)status.Value}"; + + if (!string.IsNullOrEmpty(type)) + filter += $" and WorkflowDefinitionId eq '{EscapeOData(type)}'"; + + if (createdFrom.HasValue) + filter += $" and CreateTime ge datetime'{SafeToUtc(createdFrom.Value):yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + if (createdTo.HasValue) + filter += $" and CreateTime le datetime'{SafeToUtc(createdTo.Value):yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + var query = _workflowTable.Value.QueryAsync(filter: filter); + var entities = new List(); + + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + entities.Add(entity); + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + + return entities.Skip(skip).Take(take).Select(WorkflowTableEntity.ToInstance); + } + + public async Task CreateEvent(Event newEvent, CancellationToken cancellationToken = default) + { + newEvent.Id = Guid.NewGuid().ToString(); + var entity = EventTableEntity.FromInstance(newEvent); + await _eventTable.Value.AddEntityAsync(entity, cancellationToken); + return newEvent.Id; + } + + public async Task GetEvent(string id, CancellationToken cancellationToken = default) + { + try + { + var response = await _eventTable.Value.GetEntityAsync("event", id, cancellationToken: cancellationToken); + return EventTableEntity.ToInstance(response.Value); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return null; + } + } + + public async Task> GetRunnableEvents(DateTime asAt, CancellationToken cancellationToken = default) + { + var utcAsAt = SafeToUtc(asAt); + var query = _eventTable.Value.QueryAsync( + filter: $"PartitionKey eq 'event' and IsProcessed eq false and EventTime le datetime'{utcAsAt:yyyy-MM-ddTHH:mm:ss.fffffffZ}'", + cancellationToken: cancellationToken); + + var result = new List(); + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(cancellationToken); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + result.Add(entity.RowKey); + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + return result; + } + + public async Task> GetEvents(string eventName, string eventKey, DateTime asOf, CancellationToken cancellationToken = default) + { + var utcAsOf = SafeToUtc(asOf); + var filter = $"PartitionKey eq 'event' and EventName eq '{EscapeOData(eventName)}' and EventKey eq '{EscapeOData(eventKey)}' and EventTime ge datetime'{utcAsOf:yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + var query = _eventTable.Value.QueryAsync(filter: filter, cancellationToken: cancellationToken); + var result = new List(); + + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(cancellationToken); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + result.Add(entity.RowKey); + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + return result; + } + + public async Task MarkEventProcessed(string id, CancellationToken cancellationToken = default) + { + var entity = await _eventTable.Value.GetEntityAsync("event", id, cancellationToken: cancellationToken); + entity.Value.IsProcessed = true; + try + { + await _eventTable.Value.UpdateEntityAsync(entity.Value, entity.Value.ETag, cancellationToken: cancellationToken); + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + // Another worker already updated this entity; it is already marked processed. + } + } + + public async Task MarkEventUnprocessed(string id, CancellationToken cancellationToken = default) + { + var entity = await _eventTable.Value.GetEntityAsync("event", id, cancellationToken: cancellationToken); + entity.Value.IsProcessed = false; + try + { + await _eventTable.Value.UpdateEntityAsync(entity.Value, entity.Value.ETag, cancellationToken: cancellationToken); + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + // Concurrent update; the entity may already be in the desired state. + } + } + + public async Task CreateEventSubscription(EventSubscription subscription, CancellationToken cancellationToken = default) + { + subscription.Id = Guid.NewGuid().ToString(); + var entity = SubscriptionTableEntity.FromInstance(subscription); + await _subscriptionTable.Value.AddEntityAsync(entity, cancellationToken); + return subscription.Id; + } + + public async Task> GetSubscriptions(string eventName, string eventKey, DateTime asOf, CancellationToken cancellationToken = default) + { + var utcAsOf = SafeToUtc(asOf); + var filter = $"PartitionKey eq 'subscription' and EventName eq '{EscapeOData(eventName)}' and EventKey eq '{EscapeOData(eventKey)}' and SubscribeAsOf le datetime'{utcAsOf:yyyy-MM-ddTHH:mm:ss.fffffffZ}'"; + + var query = _subscriptionTable.Value.QueryAsync(filter: filter, cancellationToken: cancellationToken); + var result = new List(); + + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(cancellationToken); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + result.Add(SubscriptionTableEntity.ToInstance(entity)); + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + return result; + } + + public async Task TerminateSubscription(string eventSubscriptionId, CancellationToken cancellationToken = default) + { + await _subscriptionTable.Value.DeleteEntityAsync("subscription", eventSubscriptionId, cancellationToken: cancellationToken); + } + + public async Task GetSubscription(string eventSubscriptionId, CancellationToken cancellationToken = default) + { + try + { + var response = await _subscriptionTable.Value.GetEntityAsync("subscription", eventSubscriptionId, cancellationToken: cancellationToken); + return SubscriptionTableEntity.ToInstance(response.Value); + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return null; + } + } + + public async Task GetFirstOpenSubscription(string eventName, string eventKey, DateTime asOf, CancellationToken cancellationToken = default) + { + var subscriptions = await GetSubscriptions(eventName, eventKey, asOf, cancellationToken); + return subscriptions.FirstOrDefault(x => string.IsNullOrEmpty(x.ExternalToken)); + } + + public async Task SetSubscriptionToken(string eventSubscriptionId, string token, string workerId, DateTime expiry, CancellationToken cancellationToken = default) + { + try + { + var entity = await _subscriptionTable.Value.GetEntityAsync("subscription", eventSubscriptionId, cancellationToken: cancellationToken); + + if (!string.IsNullOrEmpty(entity.Value.ExternalToken)) + return false; + + entity.Value.ExternalToken = token; + entity.Value.ExternalWorkerId = workerId; + entity.Value.ExternalTokenExpiry = expiry; + + await _subscriptionTable.Value.UpdateEntityAsync(entity.Value, entity.Value.ETag, cancellationToken: cancellationToken); + return true; + } + catch (RequestFailedException ex) when (ex.Status == 404 || ex.Status == 412) + { + // 404: subscription gone; 412: another worker acquired the token concurrently. + return false; + } + } + + public async Task ClearSubscriptionToken(string eventSubscriptionId, string token, CancellationToken cancellationToken = default) + { + var entity = await _subscriptionTable.Value.GetEntityAsync("subscription", eventSubscriptionId, cancellationToken: cancellationToken); + + if (entity.Value.ExternalToken != token) + throw new InvalidOperationException(); + + entity.Value.ExternalToken = null; + entity.Value.ExternalWorkerId = null; + entity.Value.ExternalTokenExpiry = null; + + try + { + await _subscriptionTable.Value.UpdateEntityAsync(entity.Value, entity.Value.ETag, cancellationToken: cancellationToken); + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + // Concurrent update; token already cleared by another worker. + } + } + + public async Task ScheduleCommand(ScheduledCommand command) + { + // Use a deterministic RowKey derived from CommandName+Data to deduplicate commands. + // Hash to ensure the key is always a safe Azure Table Storage RowKey. + var rowKey = ComputeCommandRowKey(command.CommandName, command.Data); + var entity = ScheduledCommandTableEntity.FromInstance(command, rowKey); + await _commandTable.Value.UpsertEntityAsync(entity, TableUpdateMode.Replace); + } + + private static string ComputeCommandRowKey(string commandName, string data) + { + var input = $"{commandName}_{data}"; + using (var md5 = System.Security.Cryptography.MD5.Create()) + { + var hash = md5.ComputeHash(System.Text.Encoding.UTF8.GetBytes(input)); + return new Guid(hash).ToString("N"); + } + } + + public async Task ProcessCommands(DateTimeOffset asOf, Func action, CancellationToken cancellationToken = default) + { + var query = _commandTable.Value.QueryAsync( + filter: $"PartitionKey eq 'command' and ExecuteTime le {asOf.UtcDateTime.Ticks}", + cancellationToken: cancellationToken); + + var pages = query.AsPages(); + var enumerator = pages.GetAsyncEnumerator(cancellationToken); + try + { + while (await enumerator.MoveNextAsync()) + { + foreach (var entity in enumerator.Current.Values) + { + try + { + var command = ScheduledCommandTableEntity.ToInstance(entity); + await action(command); + await _commandTable.Value.DeleteEntityAsync(entity.PartitionKey, entity.RowKey, cancellationToken: cancellationToken); + } + catch (Exception ex) + { + _logger?.LogError(ex, "Error processing scheduled command {CommandName}", entity.CommandName); + } + } + } + } + finally + { + await enumerator.DisposeAsync(); + } + } + + public async Task PersistErrors(IEnumerable errors, CancellationToken cancellationToken = default) + { + foreach (var error in errors) + { + var entity = new TableEntity("error", Guid.NewGuid().ToString()) + { + ["WorkflowId"] = error.WorkflowId, + ["ExecutionPointerId"] = error.ExecutionPointerId, + ["ErrorTime"] = error.ErrorTime, + ["Message"] = error.Message + }; + + await _errorTable.Value.AddEntityAsync(entity, cancellationToken); + } + } + + public void EnsureStoreExists() + { + _tableServiceClient.CreateTableIfNotExists(_workflowTableName); + _tableServiceClient.CreateTableIfNotExists(_eventTableName); + _tableServiceClient.CreateTableIfNotExists(_subscriptionTableName); + _tableServiceClient.CreateTableIfNotExists(_commandTableName); + _tableServiceClient.CreateTableIfNotExists(_errorTableName); + } + } +} \ No newline at end of file diff --git a/src/providers/WorkflowCore.Providers.Azure/WorkflowCore.Providers.Azure.csproj b/src/providers/WorkflowCore.Providers.Azure/WorkflowCore.Providers.Azure.csproj index 65517764c..fb0b297c8 100644 --- a/src/providers/WorkflowCore.Providers.Azure/WorkflowCore.Providers.Azure.csproj +++ b/src/providers/WorkflowCore.Providers.Azure/WorkflowCore.Providers.Azure.csproj @@ -16,6 +16,7 @@ + diff --git a/test/Directory.Build.props b/test/Directory.Build.props index cd70cf018..e73b378ba 100644 --- a/test/Directory.Build.props +++ b/test/Directory.Build.props @@ -3,6 +3,7 @@ net6.0;net8.0 latest false + true diff --git a/test/WorkflowCore.Tests.Azure/AzureTableStorageDockerSetup.cs b/test/WorkflowCore.Tests.Azure/AzureTableStorageDockerSetup.cs new file mode 100644 index 000000000..286f9957c --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/AzureTableStorageDockerSetup.cs @@ -0,0 +1,36 @@ +using System.Threading.Tasks; +using Testcontainers.Azurite; +using Xunit; + +namespace WorkflowCore.Tests.Azure +{ + public class AzureTableStorageDockerSetup : IAsyncLifetime + { + private readonly AzuriteContainer _azuriteContainer; + + public static string ConnectionString { get; private set; } + + public AzureTableStorageDockerSetup() + { + _azuriteContainer = new AzuriteBuilder() + .WithInMemoryPersistence() + .Build(); + } + + public async Task InitializeAsync() + { + await _azuriteContainer.StartAsync(); + ConnectionString = _azuriteContainer.GetConnectionString(); + } + + public async Task DisposeAsync() + { + await _azuriteContainer.DisposeAsync(); + } + } + + [CollectionDefinition("AzureTableStorage collection")] + public class AzureTableStorageCollection : ICollectionFixture + { + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/AzureTableStoragePersistenceProviderFixture.cs b/test/WorkflowCore.Tests.Azure/AzureTableStoragePersistenceProviderFixture.cs new file mode 100644 index 000000000..5214e435a --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/AzureTableStoragePersistenceProviderFixture.cs @@ -0,0 +1,33 @@ +using Azure.Data.Tables; +using WorkflowCore.Interface; +using WorkflowCore.Providers.Azure.Services; +using WorkflowCore.UnitTests; +using Xunit; + +namespace WorkflowCore.Tests.Azure +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStoragePersistenceProviderFixture : BasePersistenceFixture + { + private IPersistenceProvider _subject; + + public AzureTableStoragePersistenceProviderFixture(AzureTableStorageDockerSetup dockerSetup) + { + } + + protected override IPersistenceProvider Subject + { + get + { + if (_subject == null) + { + var tableServiceClient = new TableServiceClient(AzureTableStorageDockerSetup.ConnectionString); + var provider = new AzureTableStoragePersistenceProvider(tableServiceClient, "TestWorkflowCore"); + provider.EnsureStoreExists(); + _subject = provider; + } + return _subject; + } + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageActivityScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageActivityScenario.cs new file mode 100644 index 000000000..f315c1e55 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageActivityScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageActivityScenario : ActivityScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestActivity")); + } + } +} diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageBasicScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageBasicScenario.cs new file mode 100644 index 000000000..4c755159c --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageBasicScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageBasicScenario : BasicScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageCompensationScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageCompensationScenario.cs new file mode 100644 index 000000000..fa326e833 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageCompensationScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageCompensationScenario : CompensationScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDataScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDataScenario.cs new file mode 100644 index 000000000..6f4dc33ee --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDataScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageDataScenario : DataIOScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDelayScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDelayScenario.cs new file mode 100644 index 000000000..ac0698dcb --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageDelayScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageDelayScenario : DelayScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageEventScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageEventScenario.cs new file mode 100644 index 000000000..22c96b2a2 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageEventScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageEventScenario : EventScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageForeachScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageForeachScenario.cs new file mode 100644 index 000000000..058586d47 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageForeachScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageForeachScenario : ForeachScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageIfScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageIfScenario.cs new file mode 100644 index 000000000..a3a2bfc8e --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageIfScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageIfScenario : IfScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageSagaScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageSagaScenario.cs new file mode 100644 index 000000000..2e9e6b3e2 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageSagaScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageSagaScenario : SagaScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageWhileScenario.cs b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageWhileScenario.cs new file mode 100644 index 000000000..a73db2332 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/Scenarios/AzureTableStorageWhileScenario.cs @@ -0,0 +1,15 @@ +using Microsoft.Extensions.DependencyInjection; +using WorkflowCore.IntegrationTests.Scenarios; +using Xunit; + +namespace WorkflowCore.Tests.Azure.Scenarios +{ + [Collection("AzureTableStorage collection")] + public class AzureTableStorageWhileScenario : WhileScenario + { + protected override void ConfigureServices(IServiceCollection services) + { + services.AddWorkflow(x => x.UseAzureTableStoragePersistence(AzureTableStorageDockerSetup.ConnectionString, "TestWorkflows")); + } + } +} \ No newline at end of file diff --git a/test/WorkflowCore.Tests.Azure/WorkflowCore.Tests.Azure.csproj b/test/WorkflowCore.Tests.Azure/WorkflowCore.Tests.Azure.csproj new file mode 100644 index 000000000..d85b317f6 --- /dev/null +++ b/test/WorkflowCore.Tests.Azure/WorkflowCore.Tests.Azure.csproj @@ -0,0 +1,17 @@ + + + + net8.0 + + + + + + + + + + + + + \ No newline at end of file