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

Added partition key for heavy read queries and also to avoid cross partition queries (#494)

Browse source 
* Optimized the code for better performance

* Corrected the summary

* Refactoring and alignments
This commit is contained in:
cvijayak 2020-12-10 17:41:21 +05:30 committed by GitHub
parent 8e3b68f987
commit 89632f8db3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 253 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbCollectionInfo.cs Normal file
View file

@ -0,0 +1,9 @@
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbCollectionInfo
{
public string Name { get; set; }
public string TenantId { get; set; }
public int OfferThroughput { get; set; }
}
}

35
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs
View file

@ -19,7 +19,7 @@ namespace Elsa.Persistence.DocumentDb
private readonly SemaphoreSlim clientInstanceLock;
private readonly DocumentDbStorageOptions options;
private readonly DocumentClient client;
private Dictionary<string, (string Name, Uri Uri)> collectionUris;
private Dictionary<string, (string Name, Uri Uri, string TenantId)> collectionInfos;
public DocumentDbStorage(IOptions<DocumentDbStorageOptions> options)
{
@ -60,7 +60,7 @@ namespace Elsa.Persistence.DocumentDb
try
{
if (collectionUris != null)
if (collectionInfos != null)
{
return client;
}
@ -72,21 +72,25 @@ namespace Elsa.Persistence.DocumentDb
Id = options.DatabaseName
});
var tasks = options.CollectionNames.Select(async collectionName =>
var tasks = options.CollectionInfos.Select(async collectionInfo =>
{
var partitionKeyDefinition = new PartitionKeyDefinition();
partitionKeyDefinition.Paths.Add("/tenantId");
var databaseUri = UriFactory.CreateDatabaseUri(database.Resource.Id);
var collection = await client.CreateDocumentCollectionIfNotExistsAsync(databaseUri, new DocumentCollection
{
Id = collectionName.Value
});
Id = collectionInfo.Value.Name,
PartitionKey = partitionKeyDefinition
}, new RequestOptions { OfferThroughput = collectionInfo.Value.OfferThroughput });
var uri = UriFactory.CreateDocumentCollectionUri(options.DatabaseName, collection.Resource.Id);
return (collectionName.Key, Name: collectionName.Value, Uri: uri);
return (collectionInfo.Key, collectionInfo.Value.Name, Uri: uri, collectionInfo.Value.TenantId);
});
var results = await Task.WhenAll(tasks);
collectionUris = results.ToDictionary(x => x.Key, x => (x.Name, x.Uri));
collectionInfos = results.ToDictionary(x => x.Key, x => (x.Name, x.Uri, x.TenantId));
return client;
}
@ -96,7 +100,20 @@ namespace Elsa.Persistence.DocumentDb
}
}
public Uri GetWorkflowDefinitionCollectionUri() => collectionUris["WorkflowDefinition"].Uri;
public Uri GetWorkflowInstanceCollectionUri() => collectionUris["WorkflowInstance"].Uri;
public async Task<(string Name, Uri Uri, string TenantId)> GetWorkflowDefinitionCollectionInfoAsync()
{
return await GetCollectionInfoAsync("WorkflowDefinition");
}
public async Task<(string Name, Uri Uri, string TenantId)> GetWorkflowInstanceCollectionInfoAsync()
{
return await GetCollectionInfoAsync("WorkflowInstance");
}
private async Task<(string Name, Uri Uri, string TenantId)> GetCollectionInfoAsync(string key)
{
await GetDocumentClient();
return collectionInfos[key];
}
}
}

7
src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs
View file

@ -4,6 +4,7 @@ using System.Collections.Generic;
namespace Elsa.Persistence.DocumentDb
{
public class DocumentDbStorageOptions
{
/// <summary>
@ -31,12 +32,12 @@ namespace Elsa.Persistence.DocumentDb
public string AuthSecret { get; set; }
/// <summary>
/// Gets or sets the collection names under the DocumentDB.
/// Gets or sets the collection details under the DocumentDB.
/// </summary>
/// <value>
/// The collection names.
/// The collection details.
/// </value>
public IDictionary<string, string> CollectionNames { get; set; }
public IDictionary<string, DocumentDbCollectionInfo> CollectionInfos { get; set; }
/// <summary>
/// Get or sets the request timeout for DocumentDB client. Default value set to 30 seconds

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

@ -6,7 +6,8 @@ namespace Elsa.Persistence.DocumentDb.Documents
{
public class WorkflowDefinitionVersionDocument : DocumentBase
{
[JsonProperty(PropertyName = "type")] public string Type { get; } = nameof(WorkflowDefinitionVersionDocument);
[JsonProperty(PropertyName = "type")]
public string Type { get; } = nameof(WorkflowDefinitionVersionDocument);
[JsonProperty(PropertyName = "definitionId")]
public string DefinitionId { get; set; }
@ -39,5 +40,8 @@ namespace Elsa.Persistence.DocumentDb.Documents
[JsonProperty(PropertyName = "isLatest")]
public bool IsLatest { get; set; }
[JsonProperty(PropertyName = "tenantId")]
public string TenantId { get; set; }
}
}

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

@ -10,7 +10,8 @@ namespace Elsa.Persistence.DocumentDb.Documents
[JsonProperty(PropertyName = "definitionId")]
public string DefinitionId { get; set; }
[JsonProperty(PropertyName = "type")] public string Type { get; } = nameof(WorkflowInstanceDocument);
[JsonProperty(PropertyName = "type")]
public string Type { get; } = nameof(WorkflowInstanceDocument);
[JsonProperty(PropertyName = "version")]
public int Version { get; set; }
@ -43,7 +44,8 @@ namespace Elsa.Persistence.DocumentDb.Documents
[JsonProperty(PropertyName = "scope")]
public WorkflowExecutionScope Scope { get; set; }
[JsonProperty(PropertyName = "input")] public Variables Input { get; set; }
[JsonProperty(PropertyName = "input")]
public Variables Input { get; set; }
[JsonProperty(PropertyName = "blockingActivities")]
public HashSet<BlockingActivity> BlockingActivities { get; set; }
@ -51,6 +53,10 @@ namespace Elsa.Persistence.DocumentDb.Documents
[JsonProperty(PropertyName = "executionLog")]
public ICollection<LogEntry> ExecutionLog { get; set; }
[JsonProperty(PropertyName = "fault")] public WorkflowFault Fault { get; set; }
[JsonProperty(PropertyName = "fault")]
public WorkflowFault Fault { get; set; }
[JsonProperty(PropertyName = "tenantId")]
public string TenantId { get; set; }
}
}

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

@ -27,12 +27,12 @@ namespace Elsa.Persistence.DocumentDb.Helpers
bool disableAutomaticIdGeneration = false,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async() => await client.CreateDocumentAsync(
return await ExecuteWithRetries(async() => await client.CreateDocumentAsync(
documentCollectionUri,
document,
options,
disableAutomaticIdGeneration,
cancellationToken));
cancellationToken), cancellationToken);
}
/// <summary>
@ -50,10 +50,10 @@ namespace Elsa.Persistence.DocumentDb.Helpers
RequestOptions options = null,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async() => await client.ReadDocumentAsync<T>(
return await ExecuteWithRetries(async() => await client.ReadDocumentAsync<T>(
documentUri,
options,
cancellationToken));
cancellationToken), cancellationToken);
}
/// <summary>
@ -73,12 +73,12 @@ namespace Elsa.Persistence.DocumentDb.Helpers
bool disableAutomaticIdGeneration = false,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async () => await client.UpsertDocumentAsync(
return await ExecuteWithRetries(async () => await client.UpsertDocumentAsync(
documentCollectionUri,
document,
options,
disableAutomaticIdGeneration,
cancellationToken));
cancellationToken), cancellationToken);
}
/// <summary>
@ -94,10 +94,10 @@ namespace Elsa.Persistence.DocumentDb.Helpers
RequestOptions options = null,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async () => await client.DeleteDocumentAsync(
return await ExecuteWithRetries(async () => await client.DeleteDocumentAsync(
documentUri,
options,
cancellationToken));
cancellationToken), cancellationToken);
}
/// <summary>
@ -116,11 +116,11 @@ namespace Elsa.Persistence.DocumentDb.Helpers
RequestOptions options = null,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async () => await client.ReplaceDocumentAsync(
return await ExecuteWithRetries(async () => await client.ReplaceDocumentAsync(
documentUri,
document,
options,
cancellationToken));
cancellationToken), cancellationToken);
}
/// <summary>
@ -134,17 +134,19 @@ namespace Elsa.Persistence.DocumentDb.Helpers
internal static async Task<StoredProcedureResponse<T>> ExecuteStoredProcedureWithRetriesAsync<T>(
this DocumentClient client,
Uri storedProcedureUri,
params object[] procedureParams)
object[] procedureParams,
CancellationToken cancellationToken = default)
{
return await client.ExecuteWithRetries(async () => await client.ExecuteStoredProcedureAsync<T>(
return await ExecuteWithRetries(async () => await client.ExecuteStoredProcedureAsync<T>(
storedProcedureUri,
procedureParams));
procedureParams), cancellationToken);
}
/// <summary>
/// Execute the function with retries on throttle
/// </summary>
internal static async Task<FeedResponse<T>> ExecuteNextWithRetriesAsync<T>(this IDocumentQuery<T> query)
internal static async Task<FeedResponse<T>> ExecuteNextWithRetriesAsync<T>(this IDocumentQuery<T> query,
CancellationToken cancellationToken = default)
{
while (true)
{
@ -152,7 +154,7 @@ namespace Elsa.Persistence.DocumentDb.Helpers
try
{
return await query.ExecuteNextAsync<T>();
return await query.ExecuteNextAsync<T>(cancellationToken);
}
catch (DocumentClientException ex) when (ex.StatusCode != null && (int) ex.StatusCode == 429)
{
@ -164,16 +166,14 @@ namespace Elsa.Persistence.DocumentDb.Helpers
timeSpan = de.RetryAfter;
}
await Task.Delay(timeSpan);
await Task.Delay(timeSpan, cancellationToken);
}
}
/// <summary>
/// Execute the function with retries on throttle
/// </summary>
internal static async Task<T> ExecuteWithRetries<T>(
this DocumentClient client,
Func<Task<T>> function)
private static async Task<T> ExecuteWithRetries<T>(Func<Task<T>> function, CancellationToken cancellationToken)
{
while (true)
{
@ -193,7 +193,7 @@ namespace Elsa.Persistence.DocumentDb.Helpers
timeSpan = de.RetryAfter;
}
await Task.Delay(timeSpan);
await Task.Delay(timeSpan, cancellationToken);
}
}
}

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

@ -1,20 +1,21 @@
using Microsoft.Azure.Documents.Linq;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
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)
internal static async Task<IList<T>> ToQueryResultAsync<T>(this IQueryable<T> source, CancellationToken cancellationToken = default)
{
var query = source.AsDocumentQuery();
var results = new List<T>();
while (query.HasMoreResults)
{
var nextResults = await query.ExecuteNextWithRetriesAsync();
var nextResults = await query.ExecuteNextWithRetriesAsync(cancellationToken);
results.AddRange(nextResults);
}

6
src/persistence/Elsa.Persistence.DocumentDb/IDocumentDbStorage.cs
View file

@ -1,4 +1,4 @@
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Client;
using System;
using System.Threading.Tasks;
@ -8,7 +8,7 @@ namespace Elsa.Persistence.DocumentDb
{
string ToString();
Task<DocumentClient> GetDocumentClient();
Uri GetWorkflowDefinitionCollectionUri();
Uri GetWorkflowInstanceCollectionUri();
Task<(string Name, Uri Uri, string TenantId)> GetWorkflowDefinitionCollectionInfoAsync();
Task<(string Name, Uri Uri, string TenantId)> GetWorkflowInstanceCollectionInfoAsync();
}
}

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

@ -8,6 +8,7 @@ using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Extensions;
using Elsa.Persistence.DocumentDb.Helpers;
using Elsa.Services;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
namespace Elsa.Persistence.DocumentDb.Services
@ -24,69 +25,113 @@ namespace Elsa.Persistence.DocumentDb.Services
}
private async Task<DocumentClient> GetDocumentClient() => await storage.GetDocumentClient();
private Uri GetCollectionUri() => storage.GetWorkflowDefinitionCollectionUri();
private async Task<(string Name, Uri Uri, string TenantId)> GetCollectionInfoAsync() =>
await storage.GetWorkflowDefinitionCollectionInfoAsync();
public async Task<WorkflowDefinitionVersion> AddAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = await GetDocumentClient();
await client.CreateDocumentWithRetriesAsync(GetCollectionUri(), document, cancellationToken: cancellationToken);
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var requestOptions = new RequestOptions
{
PartitionKey = new PartitionKey(tenantId)
};
document.TenantId = tenantId;
var response = await client.CreateDocumentWithRetriesAsync(uri, document, requestOptions, cancellationToken: cancellationToken);
document = (dynamic)response.Resource;
return Map(document);
}
public async Task<int> DeleteAsync(string id, CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var records = await client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(GetCollectionUri()).Where(c => c.DefinitionId == id).ToQueryResultAsync();
foreach (var record in records)
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
await client.DeleteDocumentAsync(record.SelfLink, cancellationToken: cancellationToken);
}
return records.Count;
PartitionKey = new PartitionKey(tenantId)
};
var requestOptions = new RequestOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.DefinitionId == id);
var documents = await query.ToQueryResultAsync(cancellationToken);
var tasks = documents.Select(d =>
{
var documentUri = new Uri(d.SelfLink, UriKind.Relative);
return client.DeleteDocumentWithRetriesAsync(documentUri,
requestOptions,
cancellationToken);
}).ToArray();
await Task.WhenAll(tasks);
return documents.Count;
}
public async Task<WorkflowDefinitionVersion> GetByIdAsync(string id, VersionOptions version, CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(GetCollectionUri())
.Where(c => c.DefinitionId == id).WithVersion(version);
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.DefinitionId == id)
.WithVersion(version);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<IEnumerable<WorkflowDefinitionVersion>> ListAsync(VersionOptions version, CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(GetCollectionUri())
.WithVersion(version).ToList();
return mapper.Map<IEnumerable<WorkflowDefinitionVersion>>(query);
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowDefinitionVersionDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.WithVersion(version);
var documents = await query.ToQueryResultAsync(cancellationToken);
return Map(documents);
}
public async Task<WorkflowDefinitionVersion> SaveAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = await GetDocumentClient();
await client.UpsertDocumentWithRetriesAsync(GetCollectionUri(), document, cancellationToken: cancellationToken);
return definition;
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var requestOptions = new RequestOptions
{
PartitionKey = new PartitionKey(tenantId)
};
document.TenantId = tenantId;
var response = await client.UpsertDocumentWithRetriesAsync(uri, document, requestOptions, cancellationToken: cancellationToken);
document = (dynamic)response.Resource;
return Map(document);
}
public async Task<WorkflowDefinitionVersion> UpdateAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default)
{
var document = Map(definition);
var client = await GetDocumentClient();
await client.UpsertDocumentWithRetriesAsync(GetCollectionUri(), document, cancellationToken: cancellationToken);
return Map(document);
return await SaveAsync(definition, cancellationToken);
}
private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source)
{
return mapper.Map<WorkflowDefinitionVersionDocument>(source);
}
private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source)
{
return mapper.Map<WorkflowDefinitionVersion>(source);
}
private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source) => mapper.Map<WorkflowDefinitionVersionDocument>(source);
private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source) => mapper.Map<WorkflowDefinitionVersion>(source);
private IEnumerable<WorkflowDefinitionVersion> Map(IEnumerable<WorkflowDefinitionVersionDocument> source) =>
mapper.Map<IEnumerable<WorkflowDefinitionVersion>>(source);
}
}

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

@ -8,6 +8,7 @@ using Elsa.Models;
using Elsa.Persistence.DocumentDb.Documents;
using Elsa.Persistence.DocumentDb.Helpers;
using Elsa.Services;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
namespace Elsa.Persistence.DocumentDb.Services
@ -24,52 +25,87 @@ namespace Elsa.Persistence.DocumentDb.Services
}
private async Task<DocumentClient> GetDocumentClient() => await storage.GetDocumentClient();
private Uri GetCollectionUri() => storage.GetWorkflowInstanceCollectionUri();
private async Task<(string Name, Uri Uri, string TenantId)> GetCollectionInfoAsync() =>
await storage.GetWorkflowInstanceCollectionInfoAsync();
public async Task DeleteAsync(
string id,
public async Task DeleteAsync(string id,
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var requestOptions = new RequestOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.Id == id);
var document = query.AsEnumerable().FirstOrDefault();
if (document != null)
{
var documentUri = new Uri(document.SelfLink, UriKind.Relative);
await client.DeleteDocumentAsync(documentUri, cancellationToken: cancellationToken);
await client.DeleteDocumentWithRetriesAsync(documentUri,
requestOptions,
cancellationToken);
}
}
public async Task<WorkflowInstance> GetByCorrelationIdAsync(
string correlationId,
public async Task<WorkflowInstance> GetByCorrelationIdAsync(string correlationId,
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.CorrelationId == correlationId);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<WorkflowInstance> GetByIdAsync(
string id,
public async Task<WorkflowInstance> GetByIdAsync(string id,
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.Id == id);
var document = query.AsEnumerable().FirstOrDefault();
return Map(document);
}
public async Task<IEnumerable<WorkflowInstance>> ListAllAsync(CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.OrderByDescending(x => x.CreatedAt);
return mapper.Map<IEnumerable<WorkflowInstance>>(query);
var documents = await query.ToQueryResultAsync(cancellationToken);
return Map(documents);
}
public async Task<IEnumerable<(WorkflowInstance, ActivityInstance)>> ListByBlockingActivityAsync(
@ -78,9 +114,14 @@ namespace Elsa.Persistence.DocumentDb.Services
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(x => x.Status == WorkflowStatus.Executing);
if (!string.IsNullOrWhiteSpace(correlationId))
@ -91,8 +132,10 @@ namespace Elsa.Persistence.DocumentDb.Services
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);
var documents = await query.ToQueryResultAsync(cancellationToken);
var instances = Map(documents);
return instances.GetBlockingActivities(activityType).ToArray();
}
public async Task<IEnumerable<WorkflowInstance>> ListByDefinitionAsync(
@ -100,10 +143,19 @@ namespace Elsa.Persistence.DocumentDb.Services
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client
.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.DefinitionId == definitionId)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
var documents = await query.ToQueryResultAsync(cancellationToken);
return Map(documents);
}
public async Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(
@ -112,10 +164,18 @@ namespace Elsa.Persistence.DocumentDb.Services
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.DefinitionId == definitionId && c.Status == status)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
var documents = await query.ToQueryResultAsync(cancellationToken);
return Map(documents);
}
public async Task<IEnumerable<WorkflowInstance>> ListByStatusAsync(
@ -123,10 +183,18 @@ namespace Elsa.Persistence.DocumentDb.Services
CancellationToken cancellationToken = default)
{
var client = await GetDocumentClient();
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(GetCollectionUri())
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var feedOptions = new FeedOptions
{
PartitionKey = new PartitionKey(tenantId)
};
var query = client.CreateDocumentQuery<WorkflowInstanceDocument>(uri, feedOptions)
.Where(c => c.TenantId == tenantId)
.Where(c => c.Status == status)
.OrderByDescending(x => x.CreatedAt);
return Map(query.ToList());
var documents = await query.ToQueryResultAsync(cancellationToken);
return Map(documents);
}
public async Task<WorkflowInstance> SaveAsync(
@ -135,9 +203,16 @@ namespace Elsa.Persistence.DocumentDb.Services
{
var document = Map(instance);
var client = await GetDocumentClient();
var (_, uri, tenantId) = await GetCollectionInfoAsync();
var requestOptions = new RequestOptions
{
PartitionKey = new PartitionKey(tenantId)
};
document.TenantId = tenantId;
var response = await client.UpsertDocumentWithRetriesAsync(
GetCollectionUri(),
uri,
document,
requestOptions,
cancellationToken: cancellationToken);
document = (dynamic)response.Resource;
@ -146,8 +221,7 @@ namespace Elsa.Persistence.DocumentDb.Services
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) =>
private IEnumerable<WorkflowInstance> Map(IEnumerable<WorkflowInstanceDocument> source) =>
mapper.Map<IEnumerable<WorkflowInstance>>(source);
}
}