SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Added in DocumentDb Persistence layer (#79)

Browse source
This commit is contained in:
Jonathan Ruckert 2019-09-25 21:13:42 +09:30 committed by Sipke Schoorstra
parent 15278ef5a4
commit ee9b1ae649
14 changed files with 752 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
Elsa.sln
View file

@ -19,35 +19,37 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Abstractions", "src\co
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "core", "core", "{35F44BE9-13D0-417C-A01A-F0787BEE6DC3}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Console", "src\activities\Elsa.Activities.Console\Elsa.Activities.Console.csproj", "{B98B80CB-D3A3-437C-A61C-ED82C8C77B2F}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Console", "src\activities\Elsa.Activities.Console\Elsa.Activities.Console.csproj", "{B98B80CB-D3A3-437C-A61C-ED82C8C77B2F}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Http", "src\activities\Elsa.Activities.Http\Elsa.Activities.Http.csproj", "{1D63E1B4-2386-4BEC-9090-C3F3CBF25703}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Http", "src\activities\Elsa.Activities.Http\Elsa.Activities.Http.csproj", "{1D63E1B4-2386-4BEC-9090-C3F3CBF25703}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "activities", "activities", "{B43B546E-23F3-46E8-ACB7-D04F05CDA180}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Email", "src\activities\Elsa.Activities.Email\Elsa.Activities.Email.csproj", "{D20FCB88-9DCA-49EA-9CC2-5B94DD935D30}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Email", "src\activities\Elsa.Activities.Email\Elsa.Activities.Email.csproj", "{D20FCB88-9DCA-49EA-9CC2-5B94DD935D30}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Timers", "src\activities\Elsa.Activities.Timers\Elsa.Activities.Timers.csproj", "{E4B71DC4-3E73-49C3-9B4E-CA13909F222D}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Timers", "src\activities\Elsa.Activities.Timers\Elsa.Activities.Timers.csproj", "{E4B71DC4-3E73-49C3-9B4E-CA13909F222D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.MassTransit", "src\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj", "{C4939482-9447-47D1-B6AF-E9F8C6323841}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.MassTransit", "src\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj", "{C4939482-9447-47D1-B6AF-E9F8C6323841}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "persistence", "persistence", "{76710230-AC6D-4E54-9A06-15F34C155118}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.YesSql", "src\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj", "{4DB0741F-EAD9-4E37-862D-A1DA12462B7D}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.YesSql", "src\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj", "{4DB0741F-EAD9-4E37-862D-A1DA12462B7D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.UserTask", "src\activities\Elsa.Activities.UserTask\Elsa.Activities.UserTask.csproj", "{BC0F57E2-AB64-4026-B6E3-E03D9133DB26}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.UserTask", "src\activities\Elsa.Activities.UserTask\Elsa.Activities.UserTask.csproj", "{BC0F57E2-AB64-4026-B6E3-E03D9133DB26}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.AutoMapper.Extensions", "src\core\Elsa.AutoMapper.Extensions\Elsa.AutoMapper.Extensions.csproj", "{6CEF6AFB-635D-4FB4-B9EC-0806DD95E1BC}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.AutoMapper.Extensions", "src\core\Elsa.AutoMapper.Extensions\Elsa.AutoMapper.Extensions.csproj", "{6CEF6AFB-635D-4FB4-B9EC-0806DD95E1BC}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.EntityFrameworkCore", "src\persistence\Elsa.Persistence.EntityFrameworkCore\Elsa.Persistence.EntityFrameworkCore.csproj", "{349D4C87-E17F-4F51-B493-1994BF631D37}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.EntityFrameworkCore", "src\persistence\Elsa.Persistence.EntityFrameworkCore\Elsa.Persistence.EntityFrameworkCore.csproj", "{349D4C87-E17F-4F51-B493-1994BF631D37}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "dashboard", "dashboard", "{8B87D79A-F0DE-4F88-81D3-8C8C69108ADB}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Dashboard", "src\dashboard\Elsa.Dashboard\Elsa.Dashboard.csproj", "{493FC3E3-34FE-4F8D-AF37-38A5F5541FBD}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Dashboard", "src\dashboard\Elsa.Dashboard\Elsa.Dashboard.csproj", "{493FC3E3-34FE-4F8D-AF37-38A5F5541FBD}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.MongoDb", "src\persistence\Elsa.Persistence.MongoDb\Elsa.Persistence.MongoDb.csproj", "{82CDD590-2409-4C7E-8AF4-07C7D718D4D8}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.MongoDb", "src\persistence\Elsa.Persistence.MongoDb\Elsa.Persistence.MongoDb.csproj", "{82CDD590-2409-4C7E-8AF4-07C7D718D4D8}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.WorkflowDesigner", "src\dashboard\Elsa.WorkflowDesigner\Elsa.WorkflowDesigner.csproj", "{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.WorkflowDesigner", "src\dashboard\Elsa.WorkflowDesigner\Elsa.WorkflowDesigner.csproj", "{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.DocumentDb", "src\persistence\Elsa.Persistence.DocumentDb\Elsa.Persistence.DocumentDb.csproj", "{39EBECE7-538B-431C-AAB7-901AB16E6D4A}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@ -111,6 +113,10 @@ Global
{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Debug|Any CPU.Build.0 = Debug|Any CPU
{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Release|Any CPU.ActiveCfg = Release|Any CPU
{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Release|Any CPU.Build.0 = Release|Any CPU
{39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Debug|Any CPU.Build.0 = Debug|Any CPU
{39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Release|Any CPU.ActiveCfg = Release|Any CPU
{39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -119,9 +125,9 @@ Global
{300EE2D5-54C5-46F2-AD03-BB43589EA074} = {35F44BE9-13D0-417C-A01A-F0787BEE6DC3}
{3130C574-557C-4DA2-933E-AED7A9EBE48A} = {35F44BE9-13D0-417C-A01A-F0787BEE6DC3}
{35F44BE9-13D0-417C-A01A-F0787BEE6DC3} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925}
{B43B546E-23F3-46E8-ACB7-D04F05CDA180} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925}
{B98B80CB-D3A3-437C-A61C-ED82C8C77B2F} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{1D63E1B4-2386-4BEC-9090-C3F3CBF25703} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{B43B546E-23F3-46E8-ACB7-D04F05CDA180} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925}
{D20FCB88-9DCA-49EA-9CC2-5B94DD935D30} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{E4B71DC4-3E73-49C3-9B4E-CA13909F222D} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{C4939482-9447-47D1-B6AF-E9F8C6323841} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
@ -134,6 +140,7 @@ Global
{493FC3E3-34FE-4F8D-AF37-38A5F5541FBD} = {8B87D79A-F0DE-4F88-81D3-8C8C69108ADB}
{82CDD590-2409-4C7E-8AF4-07C7D718D4D8} = {76710230-AC6D-4E54-9A06-15F34C155118}
{120E4BF9-4A42-4BB9-A0B2-9597160D6E37} = {8B87D79A-F0DE-4F88-81D3-8C8C69108ADB}
{39EBECE7-538B-431C-AAB7-901AB16E6D4A} = {76710230-AC6D-4E54-9A06-15F34C155118}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158}

80
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs Normal file
View file

@ -0,0 +1,80 @@
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Newtonsoft.Json;
using Newtonsoft.Json.Serialization;
using System;
using System.Threading.Tasks;
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbStorage
{
internal DocumentDbStorageOptions Options { get; }
internal DocumentClient Client { get; }
internal Uri CollectionUri { get; private set; }
public DocumentDbStorage(string url, string authSecret, string database, string collection, DocumentDbStorageOptions options = null)
{
Options = options ?? new DocumentDbStorageOptions();
Options.DatabaseName = database;
Options.CollectionName = collection;
JsonSerializerSettings settings = new JsonSerializerSettings
{
NullValueHandling = NullValueHandling.Ignore,
DateTimeZoneHandling = DateTimeZoneHandling.Utc,
ContractResolver = new CamelCasePropertyNamesContractResolver
{
NamingStrategy = new CamelCaseNamingStrategy(false, false)
}
};
ConnectionPolicy connectionPolicy = ConnectionPolicy.Default;
connectionPolicy.ConnectionMode = Options.ConnectionMode;
connectionPolicy.ConnectionProtocol = Options.ConnectionProtocol;
connectionPolicy.RequestTimeout = Options.RequestTimeout;
connectionPolicy.RetryOptions = new RetryOptions
{
MaxRetryWaitTimeInSeconds = 10,
MaxRetryAttemptsOnThrottledRequests = 5
};
Client = new DocumentClient(new Uri(url), authSecret, settings, connectionPolicy);
Task task = Client.OpenAsync();
Task continueTask = task.ContinueWith(t => Initialize(), TaskContinuationOptions.OnlyOnRanToCompletion);
continueTask.Wait();
}
/// <summary>
/// Return the name of the database
/// </summary>
/// <returns></returns>
public override string ToString() => $"DoucmentDb Database : {Options.DatabaseName}";
private void Initialize()
{
//_logger.Info($"Creating database : {Options.DatabaseName}");
var databaseTask = Client.CreateDatabaseIfNotExistsAsync(new Database { Id = Options.DatabaseName });
// create document collection
Task<ResourceResponse<DocumentCollection>> collectionTask = databaseTask.ContinueWith(t =>
{
//_logger.Info($"Creating document collection : {t.Result.Resource.Id}");
Uri databaseUri = UriFactory.CreateDatabaseUri(t.Result.Resource.Id);
return Client.CreateDocumentCollectionIfNotExistsAsync(databaseUri, new DocumentCollection { Id = Options.CollectionName });
}, TaskContinuationOptions.OnlyOnRanToCompletion).Unwrap();
var continueTask = collectionTask.ContinueWith(t =>
{
CollectionUri = UriFactory.CreateDocumentCollectionUri(Options.DatabaseName, t.Result.Resource.Id);
}, TaskContinuationOptions.OnlyOnRanToCompletion);
continueTask.Wait();
if (continueTask.IsFaulted || continueTask.IsCanceled)
{
throw new ApplicationException("Unable to setup the storage database", databaseTask.Exception);
}
}
}
}

68
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs Normal file
View file

@ -0,0 +1,68 @@
using Microsoft.Azure.Documents.Client;
using System;
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbStorageOptions
{
/// <summary>
/// Gets or sets the name of the database.
/// </summary>
/// <value>
/// The name of the database.
/// </value>
internal string DatabaseName { get; set; }
/// <summary>
/// Gets or sets the name of the collection.
/// </summary>
/// <value>
/// The name of the collection.
/// </value>
internal string CollectionName { get; set; }
/// <summary>
/// Get or sets the request timeout for DocumentDB client. Default value set to 30 seconds
/// </summary>
public TimeSpan RequestTimeout { get; set; }
/// <summary>
/// Get or set the interval timespan to process expired entries. Default value 15 minutes
/// Expired items under "locks", "jobs", "lists", "sets", "hashs", "counters/aggregated" will be checked
/// </summary>
public TimeSpan ExpirationCheckInterval { get; set; }
/// <summary>
/// Get or sets the interval timespan to aggregated the counters. Default value 1 minute
/// </summary>
public TimeSpan CountersAggregateInterval { get; set; }
/// <summary>
/// Gets or sets the interval timespan to poll the queue for processing any new jobs. Default value 2 minutes
/// </summary>
public TimeSpan QueuePollInterval { get; set; }
/// <summary>
/// Gets or sets the connection mode for the DocumentDB client. Default value is Direct.
/// </summary>
public ConnectionMode ConnectionMode { get; set; }
/// <summary>
/// Gets or sets the connection protocol for the DocumentDB client. Default value is TCP.
/// </summary>
public Protocol ConnectionProtocol { get; set; }
/// <summary>
/// Create an instance of AzureDocumentDB Storage option with default values
/// </summary>
public DocumentDbStorageOptions()
{
RequestTimeout = TimeSpan.FromSeconds(30);
ExpirationCheckInterval = TimeSpan.FromMinutes(2);
CountersAggregateInterval = TimeSpan.FromMinutes(2);
QueuePollInterval = TimeSpan.FromSeconds(15);
ConnectionMode = ConnectionMode.Direct;
ConnectionProtocol = Protocol.Tcp;
}
}
}

36
src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs Normal file
View file

@ -0,0 +1,36 @@
using Elsa.Models;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Text;
namespace Elsa.Persistence.DocumentDb.Documents
{
public class WorkflowDefinitionVersionDocument
{
[JsonProperty(PropertyName = "id")]
public string Id { get; set; }
[JsonProperty(PropertyName = "type")]
public string Type { get; } = nameof(WorkflowDefinitionVersionDocument);
[JsonProperty(PropertyName = "definitionId")]
public string DefinitionId { get; set; }
[JsonProperty(PropertyName = "version")]
public int Version { get; set; }
[JsonProperty(PropertyName = "name")]
public string Name { get; set; }
[JsonProperty(PropertyName = "description")]
public string Description { get; set; }
[JsonProperty(PropertyName = "activities")]
public IList<ActivityDefinition> Activities { get; set; }
[JsonProperty(PropertyName = "connections")]
public IList<ConnectionDefinition> Connections { get; set; }
[JsonProperty(PropertyName = "variables")]
public Variables Variables { get; set; }
[JsonProperty(PropertyName = "isSingleton")]
public bool IsSingleton { get; set; }
[JsonProperty(PropertyName = "isPublished")]
public bool IsPublished { get; set; }
[JsonProperty(PropertyName = "isLatest")]
public bool IsLatest { get; set; }
}
}

45
src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs Normal file
View file

@ -0,0 +1,45 @@
using Elsa.Models;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
namespace Elsa.Persistence.DocumentDb.Documents
{
public class WorkflowInstanceDocument
{
[JsonProperty(PropertyName = "id")]
public string Id { get; set; }
[JsonProperty(PropertyName = "definitionId")]
public string DefinitionId { get; set; }
[JsonProperty(PropertyName = "type")]
public string Type { get; } = nameof(WorkflowInstanceDocument);
[JsonProperty(PropertyName = "version")]
public int Version { get; set; }
[JsonProperty(PropertyName = "status")]
public WorkflowStatus Status { get; set; }
[JsonProperty(PropertyName = "correlationId")]
public string CorrelationId { get; set; }
[JsonProperty(PropertyName = "createdAt")]
public DateTime CreatedAt { get; set; }
[JsonProperty(PropertyName = "startedAt")]
public DateTime? StartedAt { get; set; }
[JsonProperty(PropertyName = "finishedAt")]
public DateTime? FinishedAt { get; set; }
[JsonProperty(PropertyName = "faultedAt")]
public DateTime? FaultedAt { get; set; }
[JsonProperty(PropertyName = "abortedAt")]
public DateTime? AbortedAt { get; set; }
[JsonProperty(PropertyName = "activites")]
public IDictionary<string, ActivityInstance> Activities { get; set; } = new Dictionary<string, ActivityInstance>();
[JsonProperty(PropertyName = "scopes")]
public Stack<WorkflowExecutionScope> Scopes { get; set; }
[JsonProperty(PropertyName = "input")]
public Variables Input { get; set; }
[JsonProperty(PropertyName = "blockingActivities")]
public HashSet<BlockingActivity> BlockingActivities { get; set; }
[JsonProperty(PropertyName = "executionLog")]
public ICollection<LogEntry> ExecutionLog { get; set; }
[JsonProperty(PropertyName = "fault")]
public WorkflowFault Fault { get; set; }
}
}

18
src/persistence/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj Normal file
View file

@ -0,0 +1,18 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<TargetFramework>netcoreapp2.2</TargetFramework>
<LangVersion>latest</LangVersion>
<PackageVersion>1.0.0</PackageVersion>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Azure.DocumentDB.Core" Version="2.6.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
</ItemGroup>
</Project>

32
src/persistence/Elsa.Persistence.DocumentDb/Extensions/ServiceCollectionExtensions.cs Normal file
View file

@ -0,0 +1,32 @@
using Elsa.Extensions;
using Microsoft.Extensions.DependencyInjection;
using Elsa.Persistence.DocumentDb.Services;
using System;
namespace Elsa.Persistence.DocumentDb.Extensions
{
public static class ServiceCollectionExtensions
{
public static IServiceCollection UseAzureCosmosDbAsWorkflowStore(
this IServiceCollection services, string url, string authSecret, string database, string collection, DocumentDbStorageOptions options = null)
{
if (string.IsNullOrEmpty(url)) throw new ArgumentNullException(nameof(url));
if (string.IsNullOrEmpty(authSecret)) throw new ArgumentNullException(nameof(authSecret));
var storage = new DocumentDbStorage(url, authSecret, database, collection, options);
return services
.AddSingleton(storage);
}
public static IServiceCollection AddCosmosDbWorkflowInstanceStore(this IServiceCollection services)
{
return services.Replace<IWorkflowInstanceStore, CosmosDbWorkflowInstanceStore>(ServiceLifetime.Transient);
}
public static IServiceCollection AddCosmosDbWorkflowDefinitionStore(this IServiceCollection services)
{
return services.Replace<IWorkflowDefinitionStore, CosmosDbWorkflowDefinitionStore>(ServiceLifetime.Transient);
}
}
}

28
src/persistence/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs Normal file
View file

@ -0,0 +1,28 @@
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using System.Linq;
namespace Elsa.Persistence.DocumentDb.Extensions
{
public static class WorkflowDefinitionDocumentExtensions
{
public static IQueryable<WorkflowDefinitionVersionDocument> WithVersion(
this IQueryable<WorkflowDefinitionVersionDocument> query,
VersionOptions version)
{
if (version.IsDraft)
query = query.Where(x => !x.IsPublished);
if (version.IsLatest)
query = query.OrderByDescending(x => x.Version);
if (version.IsPublished)
query = query.Where(x => x.IsPublished);
if (version.IsLatestOrPublished)
query = query.Where(x => x.IsPublished || x.IsLatest);
if (version.Version > 0)
query = query.Where(x => x.Version == version.Version);
return query.OrderByDescending(x => x.Version);
}
}
}

146
src/persistence/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs Normal file
View file

@ -0,0 +1,146 @@
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Linq;
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Elsa.Persistence.DocumentDb.Helpers
{
internal static class ClientHelper
{
/// <summary>
/// Creates a document as an asynchronous operation in the Azure Cosmos DB service.
/// </summary>
/// <param name="client"></param>
/// <param name="documentCollectionUri">the URI of the document collection to create the document in.</param>
/// <param name="document">the document object.</param>
/// <param name="options">The request options for the request.</param>
/// <param name="disableAutomaticIdGeneration">Disables the automatic id generation, will throw an exception if id is missing.</param>
/// <param name="cancellationToken">(Optional) <see cref="T:System.Threading.CancellationToken" /> representing request cancellation.</param>
/// <returns></returns>
internal static Task<ResourceResponse<Document>> CreateDocumentWithRetriesAsync(this DocumentClient client, Uri documentCollectionUri, object document, RequestOptions options = null, bool disableAutomaticIdGeneration = false, CancellationToken cancellationToken = default(CancellationToken))
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.CreateDocumentAsync(documentCollectionUri, document, options, disableAutomaticIdGeneration, cancellationToken)), cancellationToken);
}
/// <summary>
/// Reads a <see cref="T:Microsoft.Azure.Documents.Document" /> as a generic type T from the Azure Cosmos DB service as an asynchronous operation.
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="client"></param>
/// <param name="documentUri">A URI to the Document resource to be read.</param>
/// <param name="options">The request options for the request.</param>
/// <param name="cancellationToken">(Optional) <see cref="T:System.Threading.CancellationToken" /> representing request cancellation.</param>
/// <returns></returns>
internal static Task<DocumentResponse<T>> ReadDocumentWithRetriesAsync<T>(this DocumentClient client, Uri documentUri, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.ReadDocumentAsync<T>(documentUri, options, cancellationToken)), cancellationToken);
}
/// <summary>
/// Upserts a document as an asynchronous operation in the Azure Cosmos DB service.
/// </summary>
/// <param name="client"></param>
/// <param name="documentCollectionUri">the URI of the document collection to upsert the document in.</param>
/// <param name="document">the document object.</param>
/// <param name="options">The request options for the request.</param>
/// <param name="disableAutomaticIdGeneration">Disables the automatic id generation, will throw an exception if id is missing.</param>
/// <param name="cancellationToken">(Optional) <see cref="T:System.Threading.CancellationToken" /> representing request cancellation.</param>
internal static Task<ResourceResponse<Document>> UpsertDocumentWithRetriesAsync(this DocumentClient client, Uri documentCollectionUri, object document, RequestOptions options = null, bool disableAutomaticIdGeneration = false, CancellationToken cancellationToken = default(CancellationToken))
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.UpsertDocumentAsync(documentCollectionUri, document, options, disableAutomaticIdGeneration, cancellationToken)), cancellationToken);
}
/// <summary>
/// Delete a document as an asynchronous operation from the Azure Cosmos DB service.
/// </summary>
/// <param name="client"></param>
/// <param name="documentUri">the URI of the document to delete.</param>
/// <param name="options">The request options for the request.</param>
/// <param name="cancellationToken">(Optional) <see cref="T:System.Threading.CancellationToken" /> representing request cancellation.</param>
internal static Task<ResourceResponse<Document>> DeleteDocumentWithRetriesAsync(this DocumentClient client, Uri documentUri, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.DeleteDocumentAsync(documentUri, options, cancellationToken)), cancellationToken);
}
/// <summary>
/// Replaces a document as an asynchronous operation in the Azure Cosmos DB service.
/// </summary>
/// <param name="client"></param>
/// <param name="documentUri">the URI of the document to be updated.</param>
/// <param name="document">the updated document.</param>
/// <param name="options">The request options for the request.</param>
/// <param name="cancellationToken">(Optional) <see cref="T:System.Threading.CancellationToken" /> representing request cancellation.</param>
/// <returns></returns>
internal static Task<ResourceResponse<Document>> ReplaceDocumentWithRetriesAsync(this DocumentClient client, Uri documentUri, object document, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.ReplaceDocumentAsync(documentUri, document, options, cancellationToken)), cancellationToken);
}
/// <summary>
/// Executes a stored procedure against a collection as an asynchronous operation from the Azure Cosmos DB service.
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="client"></param>
/// <param name="storedProcedureUri">the URI of the stored procedure to be executed.</param>
/// <param name="procedureParams">the parameters for the stored procedure execution.</param>
/// <returns></returns>
internal static Task<StoredProcedureResponse<T>> ExecuteStoredProcedureWithRetriesAsync<T>(this DocumentClient client, Uri storedProcedureUri, params object[] procedureParams)
{
return Task.Run(async () => await client.ExecuteWithRetries(() => client.ExecuteStoredProcedureAsync<T>(storedProcedureUri, procedureParams)));
}
/// <summary>
/// Execute the function with retries on throttle
/// </summary>
internal static async Task<FeedResponse<T>> ExecuteNextWithRetriesAsync<T>(this IDocumentQuery<T> query)
{
while (true)
{
TimeSpan timeSpan;
try
{
return await query.ExecuteNextAsync<T>();
}
catch (DocumentClientException ex) when (ex.StatusCode != null && (int)ex.StatusCode == 429)
{
timeSpan = ex.RetryAfter;
}
catch (AggregateException ex) when (ex.InnerException is DocumentClientException de && de.StatusCode != null && (int)de.StatusCode == 429)
{
timeSpan = de.RetryAfter;
}
await Task.Delay(timeSpan);
}
}
/// <summary>
/// Execute the function with retries on throttle
/// </summary>
internal static async Task<T> ExecuteWithRetries<T>(this DocumentClient client, Func<Task<T>> function)
{
while (true)
{
TimeSpan timeSpan;
try
{
return await function();
}
catch (DocumentClientException ex) when (ex.StatusCode != null && (int)ex.StatusCode == 429)
{
timeSpan = ex.RetryAfter;
}
catch (AggregateException ex) when (ex.InnerException is DocumentClientException de && de.StatusCode != null && (int)de.StatusCode == 429)
{
timeSpan = de.RetryAfter;
}
await Task.Delay(timeSpan);
}
}
}
}

28
src/persistence/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs Normal file
View file

@ -0,0 +1,28 @@
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Linq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Elsa.Persistence.DocumentDb.Helpers
{
internal static class QueryHelper
{
internal static List<T> ToQueryResult<T>(this IQueryable<T> source)
{
IDocumentQuery<T> query = source.AsDocumentQuery();
List<T> results = new List<T>();
while (query.HasMoreResults)
{
Task<FeedResponse<T>> task = Task.Run(async () => await query.ExecuteNextWithRetriesAsync());
task.Wait();
results.AddRange(task.Result);
}
return results;
}
}
}

28
src/persistence/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs Normal file
View file

@ -0,0 +1,28 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Text;
namespace Elsa.Persistence.DocumentDb.Helpers
{
internal static class TimeHelper
{
private static readonly DateTime epochDateTime = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
internal static int ToEpoch(this DateTime date)
{
if (date.Equals(DateTime.MinValue)) return int.MinValue;
TimeSpan epochTimeSpan = date - epochDateTime;
return (int)epochTimeSpan.TotalSeconds;
}
internal static DateTime ToDateTime(this int totalSeconds) => epochDateTime.AddSeconds(totalSeconds);
internal static string TryParseToEpoch(this string s)
{
return DateTime.TryParse(s, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date)
? date.ToEpoch().ToString(CultureInfo.InvariantCulture)
: s;
}
}
}

15
src/persistence/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs Normal file
View file

@ -0,0 +1,15 @@
using AutoMapper;
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
namespace Elsa.Persistence.DocumentDb.Mapping
{
public class DocumentProfile : Profile
{
public DocumentProfile()
{
CreateMap<WorkflowDefinitionVersion, WorkflowDefinitionVersionDocument>().ReverseMap();
CreateMap<WorkflowInstance, WorkflowInstanceDocument>().ReverseMap();
}
}
}

94
src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs Normal file
View file

@ -0,0 +1,94 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;
using Elsa.AutoMapper.Extensions.NodaTime;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Extensions;
using Elsa.Persistence.DocumentDb.Helpers;
using Elsa.Persistence.DocumentDb.Mapping;
namespace Elsa.Persistence.DocumentDb.Services
{
public class CosmosDbWorkflowDefinitionStore : IWorkflowDefinitionStore
{
readonly IMapper _mapper;
readonly DocumentDbStorage _storage;
public CosmosDbWorkflowDefinitionStore(DocumentDbStorage storage)
{
_storage = storage;
var configuration = new MapperConfiguration(cfg =>
{
cfg.AddProfile<InstantProfile>();
cfg.AddProfile<DocumentProfile>();
});
_mapper = configuration.CreateMapper();
}
public async Task AddAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = _storage.Client;
await client.CreateDocumentWithRetriesAsync(_storage.CollectionUri, document);
}
public async Task<int> DeleteAsync(string id, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var records = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(_storage.CollectionUri).Where(c => c.DefinitionId == id).ToQueryResult();
foreach (var record in records)
{
await client.DeleteDocumentAsync(record.Id);
}
return records.Count;
}
public Task<WorkflowDefinitionVersion> GetByIdAsync(string id, VersionOptions version, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(_storage.CollectionUri)
.Where(c => c.DefinitionId == id).WithVersion(version);
var document = query.AsEnumerable().FirstOrDefault();
return Task.FromResult(Map(document));
}
public Task<IEnumerable<WorkflowDefinitionVersion>> ListAsync(VersionOptions version, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(_storage.CollectionUri)
.WithVersion(version).ToList();
return Task.FromResult(_mapper.Map<IEnumerable<WorkflowDefinitionVersion>>(query));
}
public async Task<WorkflowDefinitionVersion> SaveAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = _storage.Client;
await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document);
return definition;
}
public async Task<WorkflowDefinitionVersion> UpdateAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = _storage.Client;
await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document);
return Map(document);
}
private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source)
{
return _mapper.Map<WorkflowDefinitionVersionDocument>(source);
}
private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source)
{
return _mapper.Map<WorkflowDefinitionVersion>(source);
}
}
}

114
src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs Normal file
View file

@ -0,0 +1,114 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;
using Elsa.AutoMapper.Extensions.NodaTime;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Helpers;
using Elsa.Persistence.DocumentDb.Mapping;
namespace Elsa.Persistence.DocumentDb.Services
{
public class CosmosDbWorkflowInstanceStore : IWorkflowInstanceStore
{
readonly IMapper _mapper;
readonly DocumentDbStorage _storage;
public CosmosDbWorkflowInstanceStore(DocumentDbStorage storage)
{
var configuration = new MapperConfiguration(cfg =>
{
cfg.AddProfile<InstantProfile>();
cfg.AddProfile<DocumentProfile>();
});
_mapper = configuration.CreateMapper();
_storage = storage;
}
public async Task DeleteAsync(string id, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.Id == id);
await client.DeleteDocumentAsync(id);
}
public Task<WorkflowInstance> GetByCorrelationIdAsync(string correlationId, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.CorrelationId == correlationId);
var document = query.AsEnumerable().FirstOrDefault();
return Task.FromResult(Map(document));
}
public Task<WorkflowInstance> GetByIdAsync(string id, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.Id == id);
var document = query.AsEnumerable().FirstOrDefault();
return Task.FromResult(Map(document));
}
public Task<IEnumerable<WorkflowInstance>> ListAllAsync(CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri);
return Task.FromResult(_mapper.Map<IEnumerable<WorkflowInstance>>(query));
}
public Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(string activityType, string correlationId = null, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri).Where(x => x.Status == WorkflowStatus.Executing);
if (!string.IsNullOrWhiteSpace(correlationId))
{
query = query.Where(x => x.CorrelationId == correlationId);
}
query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType));
var instances = Map(query.ToList());
return Task.FromResult(instances.GetBlockingActivities(activityType));
}
public Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(string definitionId, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.DefinitionId == definitionId);
return Task.FromResult(Map(query.ToList()));
}
public Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(string definitionId, WorkflowStatus status, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.DefinitionId == definitionId && c.Status == status);
return Task.FromResult(Map(query.ToList()));
}
public Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken = default)
{
var client = _storage.Client;
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(_storage.CollectionUri)
.Where(c => c.Status == status);
return Task.FromResult(Map(query.ToList()));
}
public async Task SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default)
{
var document = Map(instance);
var client = _storage.Client;
await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document);
}
private WorkflowInstanceDocument Map(WorkflowInstance source) => _mapper.Map<WorkflowInstanceDocument>(source);
private WorkflowInstance Map(WorkflowInstanceDocument source) => _mapper.Map<WorkflowInstance>(source);
private IEnumerable<WorkflowInstance> Map(IEnumerable<WorkflowInstanceDocument> source) =>
_mapper.Map<IEnumerable<WorkflowInstance>>(source);
}
}