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

Remove DocumentDB provider

Browse source
This commit is contained in:
Sipke Schoorstra 2020-10-06 15:13:02 +02:00
parent 9394f419b9
commit bd46f80401
14 changed files with 0 additions and 896 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs
View file

@ -1,57 +0,0 @@
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Newtonsoft.Json;
using Newtonsoft.Json.Serialization;
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbStorage
{
private DocumentDbStorageOptions Options { get; }
internal DocumentClient Client { get; }
public DocumentDbStorage(DocumentDbStorageOptions options)
{
Options = options;
var settings = new JsonSerializerSettings
{
NullValueHandling = NullValueHandling.Ignore,
DateTimeZoneHandling = DateTimeZoneHandling.Utc,
ContractResolver = new CamelCasePropertyNamesContractResolver
{
NamingStrategy = new CamelCaseNamingStrategy(false, false)
}
};
var 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(options.Url, options.Secret, settings, connectionPolicy);
}
public override string ToString() => $"DocumentDb Database: {Options.DatabaseName}";
public async Task<Uri> GetCollectionAsync(string collectionName, CancellationToken cancellationToken = default)
{
var database = await Client.CreateDatabaseIfNotExistsAsync(new Database { Id = Options.DatabaseName });
var databaseUri = UriFactory.CreateDatabaseUri(database.Resource.Id);
var collection = await Client.CreateDocumentCollectionIfNotExistsAsync(
databaseUri,
new DocumentCollection { Id = collectionName });
return UriFactory.CreateDocumentCollectionUri(Options.DatabaseName, collection.Resource.Id);
}
}
}

62
src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs
View file

@ -1,62 +0,0 @@
using Microsoft.Azure.Documents.Client;
using System;
using System.ComponentModel.DataAnnotations;
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbStorageOptions
{
[Required]
public Uri Url { get; set; }
[Required]
public string Secret { get; set; }
[Required]
public string DatabaseName { 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;
}
}
}

44
src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs
View file

@ -1,44 +0,0 @@
using Elsa.Models;
using Newtonsoft.Json;
using System.Collections.Generic;
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 = "isDisabled")]
public bool IsDisabled { get; set; }
[JsonProperty(PropertyName = "isPublished")]
public bool IsPublished { get; set; }
[JsonProperty(PropertyName = "isLatest")]
public bool IsLatest { get; set; }
}
}

61
src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs
View file

@ -1,61 +0,0 @@
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 = "activities")]
public IDictionary<string, ActivityInstance> Activities { get; set; } = new Dictionary<string, ActivityInstance>();
[JsonProperty(PropertyName = "variables")]
public Variables Variables { get; set; }
[JsonProperty(PropertyName = "input")] public Variable? Input { get; set; }
[JsonProperty(PropertyName = "input")] public Variable? Output { get; set; }
[JsonProperty(PropertyName = "blockingActivities")]
public HashSet<BlockingActivity> BlockingActivities { get; set; }
[JsonProperty(PropertyName = "scheduledActivities")]
public Stack<string> ScheduledActivities { get; set; }
[JsonProperty(PropertyName = "executionLog")]
public ICollection<ExecutionLogEntry> ExecutionLog { get; set; }
[JsonProperty(PropertyName = "fault")] public WorkflowFault Fault { get; set; }
}
}

36
src/providers/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj
View file

@ -1,36 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<LangVersion>8.0</LangVersion>
<PackageVersion>1.0.0</PackageVersion>
<Authors>Elsa Contributors</Authors>
<Description>
Elsa is a set of workflow libraries and tools that enable super-fast workflowing capabilities in any .NET Core application.
This package provides a CosmosDb persistence provider.
</Description>
<Copyright>2019</Copyright>
<PackageProjectUrl>https://github.com/elsa-workflows/elsa-core</PackageProjectUrl>
<RepositoryUrl>https://github.com/elsa-workflows/elsa-core</RepositoryUrl>
<RepositoryType>GitHub</RepositoryType>
<PackageTags>elsa, workflows, cosmosdb</PackageTags>
<PackageIcon>icon.png</PackageIcon>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<None Include="icon.png">
<Pack>True</Pack>
<PackagePath />
</None>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Azure.DocumentDB.Core" Version="2.11.6" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
</ItemGroup>
</Project>

48
src/providers/Elsa.Persistence.DocumentDb/Extensions/ElsaOptionsExtensions.cs
View file

@ -1,48 +0,0 @@
using Elsa.Persistence.DocumentDb.Services;
using Elsa.Extensions;
using Elsa.Mapping;
using Elsa.Persistence.DocumentDb.Mapping;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Persistence.DocumentDb.Extensions
{
public static class ElsaOptionsExtensions
{
public static ElsaOptions UseCosmosDbWorkflowDefinitionStore(this ElsaOptions options, DocumentDbStorageOptions dbOptions)
{
options
.AddCosmosDbProvider(dbOptions)
.UseWorkflowDefinitionStore(sp => sp.GetRequiredService<CosmosDbWorkflowDefinitionStore>());
options.Services.AddSingleton<CosmosDbWorkflowDefinitionStore>();
return options;
}
public static ElsaOptions UseCosmosDbWorkflowInstanceStore(this ElsaOptions options, DocumentDbStorageOptions dbOptions)
{
options
.AddCosmosDbProvider(dbOptions)
.UseWorkflowInstanceStore(sp => sp.GetRequiredService<CosmosDbWorkflowInstanceStore>());
options.Services.AddSingleton<IWorkflowInstanceStore, CosmosDbWorkflowInstanceStore>();
return options;
}
private static ElsaOptions AddCosmosDbProvider(
this ElsaOptions options,
DocumentDbStorageOptions documentDbOptions)
{
if (options.HasService<DocumentDbStorage>())
return options;
var storage = new DocumentDbStorage(documentDbOptions);
options.Services
.AddSingleton(storage)
.AddAutoMapperProfile<NodaTimeProfile>(ServiceLifetime.Singleton)
.AddAutoMapperProfile<DocumentProfile>(ServiceLifetime.Singleton);
return options;
}
}
}

31
src/providers/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs
View file

@ -1,31 +0,0 @@
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);
else if (version.IsLatest)
query = query.Where(x => x.IsLatest);
else if (version.IsPublished)
query = query.Where(x => x.IsPublished);
else if (version.IsLatestOrPublished)
query = query.Where(x => x.IsPublished || x.IsLatest);
else if (version.AllVersions)
{
// Nothing to filter.
}
else if (version.Version > 0)
query = query.Where(x => x.Version == version.Version);
return query.OrderByDescending(x => x.Version);
}
}
}

205
src/providers/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs
View file

@ -1,205 +0,0 @@
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)
{
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)
{
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)
{
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)
{
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)
{
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);
}
}
}
}

24
src/providers/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs
View file

@ -1,24 +0,0 @@
using Microsoft.Azure.Documents.Linq;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace Elsa.Persistence.DocumentDb.Helpers
{
internal static class QueryHelper
{
internal static async Task<IList<T>> ToQueryResultAsync<T>(this IQueryable<T> source)
{
var query = source.AsDocumentQuery();
var results = new List<T>();
while (query.HasMoreResults)
{
var nextResults = await Task.Run(async () => await query.ExecuteNextWithRetriesAsync());
results.AddRange(nextResults);
}
return results;
}
}
}

26
src/providers/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs
View file

@ -1,26 +0,0 @@
using System;
using System.Globalization;
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;
var 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 var date)
? date.ToEpoch().ToString(CultureInfo.InvariantCulture)
: s;
}
}
}

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

@ -1,15 +0,0 @@
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();
}
}
}

114
src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs
View file

@ -1,114 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Extensions;
using Elsa.Persistence.DocumentDb.Helpers;
namespace Elsa.Persistence.DocumentDb.Services
{
public class CosmosDbWorkflowDefinitionStore : IWorkflowDefinitionStore
{
private readonly IMapper mapper;
private readonly DocumentDbStorage storage;
private Uri? collectionUrl;
public CosmosDbWorkflowDefinitionStore(DocumentDbStorage storage, IMapper mapper)
{
this.storage = storage;
this.mapper = mapper;
collectionUrl = default;
}
public async Task<WorkflowDefinitionVersion> AddAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
await client.CreateDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken);
return Map(document);
}
public async Task<WorkflowDefinitionVersion> GetByIdAsync(string id, CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(collectionUrl).Where(c => c.Id == id);
var document = query.FirstOrDefault();
return Map(document);
}
public async Task<WorkflowDefinitionVersion> GetByIdAsync(string definitionId, VersionOptions version, CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(collectionUrl)
.Where(c => c.DefinitionId == definitionId).WithVersion(version);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<int> DeleteAsync(string id, CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var workflowDefinitionDocuments = await client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(collectionUrl).Where(c => c.DefinitionId == id).ToQueryResultAsync();
foreach (var record in workflowDefinitionDocuments)
{
await client.DeleteDocumentAsync(record.Id, cancellationToken: cancellationToken);
}
return workflowDefinitionDocuments.Count;
}
public async Task<IEnumerable<WorkflowDefinitionVersion>> ListAsync(VersionOptions version, CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client
.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(collectionUrl)
.WithVersion(version).ToList();
return mapper.Map<IEnumerable<WorkflowDefinitionVersion>>(query);
}
public async Task<WorkflowDefinitionVersion> SaveAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
await client.UpsertDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken);
return definition;
}
public async Task<WorkflowDefinitionVersion> UpdateAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
await client.UpsertDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken);
return Map(document);
}
private async Task<Uri> GetCollectionUriAsync(CancellationToken cancellationToken)
{
if (collectionUrl == null)
collectionUrl = await storage.GetCollectionAsync("WorkflowDefinitions", cancellationToken);
return collectionUrl;
}
private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source)
{
return mapper.Map<WorkflowDefinitionVersionDocument>(source);
}
private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source)
{
return mapper.Map<WorkflowDefinitionVersion>(source);
}
}
}

173
src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs
View file

@ -1,173 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AutoMapper;
using Elsa.Extensions;
using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Helpers;
namespace Elsa.Persistence.DocumentDb.Services
{
public class CosmosDbWorkflowInstanceStore : IWorkflowInstanceStore
{
private readonly IMapper mapper;
private readonly DocumentDbStorage storage;
private Uri? collectionUrl;
public CosmosDbWorkflowInstanceStore(DocumentDbStorage storage, IMapper mapper)
{
this.storage = storage;
this.mapper = mapper;
collectionUrl = default;
}
public async Task DeleteAsync(
string id,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
await client.DeleteDocumentAsync(id, cancellationToken: cancellationToken);
}
public async Task<WorkflowInstance> GetByCorrelationIdAsync(
string correlationId,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(c => c.CorrelationId == correlationId);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<WorkflowInstance> GetByIdAsync(
string id,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(c => c.Id == id);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<IEnumerable<WorkflowInstance>> ListAllAsync(CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.OrderByDescending(x => x.CreatedAt);
return mapper.Map<IEnumerable<WorkflowInstance>>(query);
}
public async Task<IEnumerable<(WorkflowInstance, BlockingActivity)>> ListByBlockingActivityTagAsync(
string activityType,
string tag,
string? correlationId = null,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(x => x.Status == WorkflowStatus.Suspended);
if (!string.IsNullOrWhiteSpace(correlationId))
query = query.Where(x => x.CorrelationId == correlationId);
query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType && y.Tag == tag));
query = query.OrderByDescending(x => x.CreatedAt);
var instances = Map(query.ToList());
return instances.GetBlockingActivities(activityType);
}
public async Task<IEnumerable<(WorkflowInstance, BlockingActivity)>> ListByBlockingActivityAsync(
string activityType,
string? correlationId = null,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(x => x.Status == WorkflowStatus.Suspended);
if (!string.IsNullOrWhiteSpace(correlationId))
query = query.Where(x => x.CorrelationId == correlationId);
query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType));
query = query.OrderByDescending(x => x.CreatedAt);
var instances = Map(query.ToList());
return instances.GetBlockingActivities(activityType);
}
public async Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(
string definitionId,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(c => c.DefinitionId == definitionId)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
}
public async Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(
string definitionId,
WorkflowStatus status,
CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(c => c.DefinitionId == definitionId && c.Status == status)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
}
public async Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken = default)
{
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(collectionUrl)
.Where(c => c.Status == status)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
}
public async Task<WorkflowInstance> SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default)
{
var document = Map(instance);
var client = storage.Client;
var collectionUrl = await GetCollectionUriAsync(cancellationToken);
var response = await client.UpsertDocumentWithRetriesAsync(
collectionUrl,
document,
cancellationToken: cancellationToken);
document = (dynamic)response.Resource;
return Map(document);
}
private async Task<Uri> GetCollectionUriAsync(CancellationToken cancellationToken)
{
if (collectionUrl == null)
collectionUrl = await storage.GetCollectionAsync("WorkflowInstances", cancellationToken);
return collectionUrl;
}
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);
}
}

BIN
src/providers/Elsa.Persistence.DocumentDb/icon.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 16 KiB