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

Minor improvements and bug fixes following the 3.1 release (#5168)

Browse source 
* Move DynamicActivity.cs to Activities directory

The DynamicActivity.cs file has been moved from the Models directory to the Activities directory. This reorganization aims to ensure that the file's location correctly reflects its namespace.

* Add GetOutput method in ActivityExtensions

A new GetOutput method has been added to the ActivityExtensions.cs file. This method allows the retrieval of output with a specific name from an activity. Useful for handling complex types in workflow activities.

* Add feature check and refactor dependencies in Elsa

The commit introduces a new feature check in the `Module` class and refactors the dependencies in MassTransit features. Specifically, it enables querying for a specific feature before configuring the dispatcher endpoints, increasing flexibility and control. In addition, the responsibility for creating `IEndpointChannelFormatter` has been shifted from `MassTransitWorkflowDispatcherFeature` to `MassTransitFeature`, aligning with responsibility distribution.

Fixes #5165

* Add HasFeature method to IModule interface

The IModule interface has been updated to include two methods, HasFeature<T>() and HasFeature(Type featureType). These methods are designed to check if a specific type of feature has been configured, enhancing the functionality provided by the interface.

* Add WorkflowRuntimeFeature dependency

Removed unused namespaces from WorkflowsApiFeature class and added a new dependency on WorkflowRuntimeFeature. This change enhances the code cleanliness and ensures all required dependencies are correctly linked.

* Add activity completion functionality to multiple contexts

This commit introduces multiple methods to handle activity completion across various contexts, including ActivityExecutionContext and ActivityCompletedContext. It also includes updates to bookmark serialization and the WorkflowRuntime. The resulting changes should improve handling of activity outcomes and status updates in the application flow.

* Handle null options in DefaultWorkflowRuntime

Added null-conditional operators to prevent potential NullReferenceExceptions in DefaultWorkflowRuntime. This change ensures that even if the 'options' object is null, the code will not throw an exception and will instead use default values where applicable.

* Add ElsaDbContextOptions to DbContextOptionsBuilder

A line of code is added to enable applying ElsaDbContextOptions as default in DbContextOptionsBuilder within PersistenceFeatureBase. This change specifies the use of ElsaDbContextOptions when configuring the context options, enhancing the database context setup in the EntityFrameworkCore.Common module.

* Remove whitespace in Elsa.Server.Web.csproj

This commit removes unnecessary whitespaces at the end of the ProjectReference and PackageReference elements, in the Elsa.Server.Web.csproj file. This improves the readability and alignment of the code and follows the best practice for XML file format.

* Add MongoDB to docker-compose.yml

A MongoDB service has been added to the docker-compose file. The configuration includes port mapping and volume mapping for MongoDB data storage. This allows more flexibility in our environment setup with MongoDB now being spun up automatically.

* Add collection check in MongoDbStore before bulk save

Adjusted code structure, and divided longer lines of code into smaller, multi-line chunks for better readability. This refactoring makes the underlying operations and structuring of the code more apparent, aiding in future code maintenance and understanding.

* Change target branch in packages.yml workflow

This commit modifies the Github actions workflow for packaging. The branch from which to fetch changes is now specified explicitly as 'origin/patch/3.1.1' instead of the default 'origin/main'. This adjustment is specific for package creation under certain conditions.
This commit is contained in:
Sipke Schoorstra 2024-04-02 07:41:46 +02:00 committed by GitHub
parent 93b78a1a1e
commit 9b9454403b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 329 additions and 220 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/packages.yml vendored
View file

@ -44,7 +44,7 @@ jobs:
run: |
if [[ "${{ github.ref }}" == refs/tags/* && "${{ github.event_name }}" == "release" && "${{ github.event.action }}" == "published" ]]; then
git fetch --no-tags --prune --depth=1 origin +refs/heads/*:refs/remotes/origin/*
git branch --remote --contains | grep origin/main
git branch --remote --contains | grep origin/patch/3.1.1
else
git fetch --no-tags --prune --depth=1 origin +refs/heads/*:refs/remotes/origin/*
git branch --remote --contains | grep origin/${BRANCH_NAME}

8
docker/docker-compose.yml
View file

@ -13,6 +13,13 @@ services:
ports:
- "5432:5432"
mongodb:
image: mongo:latest
ports:
- "127.0.0.1:27017:27017"
volumes:
- mongodb_data:/data/db
cockroachdb:
image: cockroachdb/cockroach:v22.1.0
command: start-single-node --insecure
@ -67,3 +74,4 @@ services:
volumes:
postgres-data:
cockroachdb-data:
mongodb_data:

80
src/bundles/Elsa.Server.Web/Elsa.Server.Web.csproj
View file

@ -8,52 +8,52 @@
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.MassTransit.AzureServiceBus\Elsa.MassTransit.AzureServiceBus.csproj" />
<ProjectReference Include="..\Elsa\Elsa.csproj" />
<ProjectReference Include="..\..\common\Elsa.DropIns\Elsa.DropIns.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Alterations.MassTransit\Elsa.Alterations.MassTransit.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Alterations\Elsa.Alterations.csproj" />
<ProjectReference Include="..\..\modules\Elsa.CSharp\Elsa.CSharp.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.SqlServer\Elsa.EntityFrameworkCore.SqlServer.csproj" />
<ProjectReference Include="..\..\modules\Elsa.MassTransit.RabbitMq\Elsa.MassTransit.RabbitMq.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Python\Elsa.Python.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Quartz.EntityFrameworkCore.Sqlite\Elsa.Quartz.EntityFrameworkCore.Sqlite.csproj" />
<ProjectReference Include="..\..\modules\Elsa.FileStorage\Elsa.FileStorage.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Dapper.Migrations\Elsa.Dapper.Migrations.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Dapper\Elsa.Dapper.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Email\Elsa.Email.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore\Elsa.EntityFrameworkCore.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Environments\Elsa.Environments.csproj" />
<ProjectReference Include="..\..\modules\Elsa.MassTransit\Elsa.MassTransit.csproj" />
<ProjectReference Include="..\..\modules\Elsa.MongoDb\Elsa.MongoDb.csproj" />
<ProjectReference Include="..\..\modules\Elsa.ProtoActor\Elsa.ProtoActor.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Identity\Elsa.Identity.csproj" />
<ProjectReference Include="..\..\modules\Elsa.ProtoActor.Cluster.AzureContainerApps\Elsa.ProtoActor.Cluster.AzureContainerApps.csproj" />
<ProjectReference Include="..\..\modules\Elsa.WorkflowProviders.BlobStorage\Elsa.WorkflowProviders.BlobStorage.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Hangfire\Elsa.Hangfire.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Http\Elsa.Http.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Labels\Elsa.Labels.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Liquid\Elsa.Liquid.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Quartz\Elsa.Quartz.csproj" />
<ProjectReference Include="..\..\modules\Elsa.WorkflowContexts\Elsa.WorkflowContexts.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Workflows.Api\Elsa.Workflows.Api.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.MassTransit.AzureServiceBus\Elsa.MassTransit.AzureServiceBus.csproj"/>
<ProjectReference Include="..\Elsa\Elsa.csproj"/>
<ProjectReference Include="..\..\common\Elsa.DropIns\Elsa.DropIns.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Alterations.MassTransit\Elsa.Alterations.MassTransit.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Alterations\Elsa.Alterations.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.CSharp\Elsa.CSharp.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.SqlServer\Elsa.EntityFrameworkCore.SqlServer.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.MassTransit.RabbitMq\Elsa.MassTransit.RabbitMq.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Python\Elsa.Python.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Quartz.EntityFrameworkCore.Sqlite\Elsa.Quartz.EntityFrameworkCore.Sqlite.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.FileStorage\Elsa.FileStorage.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Dapper.Migrations\Elsa.Dapper.Migrations.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Dapper\Elsa.Dapper.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Email\Elsa.Email.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore\Elsa.EntityFrameworkCore.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Environments\Elsa.Environments.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.MassTransit\Elsa.MassTransit.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.MongoDb\Elsa.MongoDb.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.ProtoActor\Elsa.ProtoActor.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Identity\Elsa.Identity.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.ProtoActor.Cluster.AzureContainerApps\Elsa.ProtoActor.Cluster.AzureContainerApps.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.WorkflowProviders.BlobStorage\Elsa.WorkflowProviders.BlobStorage.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Hangfire\Elsa.Hangfire.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Http\Elsa.Http.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Labels\Elsa.Labels.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Liquid\Elsa.Liquid.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Quartz\Elsa.Quartz.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.WorkflowContexts\Elsa.WorkflowContexts.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Workflows.Api\Elsa.Workflows.Api.csproj"/>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Azure.Identity" />
<PackageReference Include="Bogus" />
<PackageReference Include="DistributedLock.Postgres" />
<PackageReference Include="DistributedLock.Redis" />
<PackageReference Include="FluentStorage.Azure.Blobs" />
<PackageReference Include="Proto.Persistence.Sqlite" />
<PackageReference Include="Proto.Persistence.SqlServer" />
<PackageReference Include="Azure.Identity"/>
<PackageReference Include="Bogus"/>
<PackageReference Include="DistributedLock.Postgres"/>
<PackageReference Include="DistributedLock.Redis"/>
<PackageReference Include="FluentStorage.Azure.Blobs"/>
<PackageReference Include="Proto.Persistence.Sqlite"/>
<PackageReference Include="Proto.Persistence.SqlServer"/>
</ItemGroup>
<ItemGroup>
<Folder Include="App_Data\" />
<Folder Include="App_Data\"/>
</ItemGroup>
</Project>

12
src/common/Elsa.Features/Implementations/Module.cs
View file

@ -34,6 +34,18 @@ public class Module : IModule
/// <inheritdoc />
public IDictionary<object, object> Properties { get; } = new Dictionary<object, object>();
/// <inheritdoc />
public bool HasFeature<T>() where T : class, IFeature
{
return HasFeature(typeof(T));
}
/// <inheritdoc />
public bool HasFeature(Type featureType)
{
return _features.ContainsKey(featureType);
}
/// <inheritdoc />
public T Configure<T>(Action<T>? configure = default) where T : class, IFeature
=> Configure(module => (T)Activator.CreateInstance(typeof(T), module)!, configure);

10
src/common/Elsa.Features/Services/IModule.cs
View file

@ -17,6 +17,16 @@ public interface IModule
/// A dictionary into which features can stash away values for later use.
/// </summary>
IDictionary<object, object> Properties { get; }
/// <summary>
/// Returns true if a feature of the specified type has been configured.
/// </summary>
bool HasFeature<T>() where T : class, IFeature;
/// <summary>
/// Returns true if a feature of the specified type has been configured.
/// </summary>
bool HasFeature(Type featureType);
/// <summary>
/// Creates and configures a feature of the specified type.

1
src/modules/Elsa.EntityFrameworkCore.Common/PersistenceFeatureBase.cs
View file

@ -36,6 +36,7 @@ public abstract class PersistenceFeatureBase<TDbContext> : FeatureBase where TDb
/// Gets or sets the callback used to configure the <see cref="DbContextOptionsBuilder"/>.
/// </summary>
public Action<IServiceProvider, DbContextOptionsBuilder> DbContextOptionsBuilder = (_, options) => options
.UseElsaDbContextOptions(default)
.UseSqlite("Data Source=elsa.sqlite.db;Cache=Shared;", sqlite => sqlite
.MigrationsAssembly("Elsa.EntityFrameworkCore.Sqlite")
.MigrationsHistoryTable(ElsaDbContextBase.MigrationsHistoryTable, ElsaDbContextBase.ElsaSchema));

5
src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs
View file

@ -78,7 +78,10 @@ public class RabbitMqServiceBusFeature : FeatureBase
});
}
configurator.SetupWorkflowDispatcherEndpoints(context);
// Only configure the dispatcher endpoints if the Masstransit Workflow Dispatcher feature is enabled.
if (Module.HasFeature<MassTransitWorkflowDispatcherFeature>())
configurator.SetupWorkflowDispatcherEndpoints(context);
configurator.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("Elsa", false));
});
};

8
src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs
View file

@ -5,7 +5,9 @@ using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Elsa.MassTransit.Consumers;
using Elsa.MassTransit.Contracts;
using Elsa.MassTransit.Extensions;
using Elsa.MassTransit.Formatters;
using Elsa.MassTransit.Models;
using Elsa.MassTransit.Options;
using Elsa.MassTransit.Services;
@ -37,6 +39,11 @@ public class MassTransitFeature : FeatureBase
/// A delegate that can be set to configure MassTransit's <see cref="IBusRegistrationConfigurator"/>. Used by transport-level features such as AzureServiceBusFeature and RabbitMqServiceBusFeature.
/// </summary>
public Action<IBusRegistrationConfigurator>? BusConfigurator { get; set; }
/// <summary>
/// A factory that creates a <see cref="IEndpointChannelFormatter"/>.
/// </summary>
public Func<IServiceProvider, IEndpointChannelFormatter> ChannelQueueFormatterFactory { get; set; } = _ => new DefaultEndpointChannelFormatter();
/// <inheritdoc />
public override void Configure()
@ -48,6 +55,7 @@ public class MassTransitFeature : FeatureBase
{
var messageTypes = this.GetMessages();
Services.AddSingleton(ChannelQueueFormatterFactory);
Services.Configure<MassTransitWorkflowDispatcherOptions>(x => { });
Services.AddActivityProvider<MassTransitActivityTypeProvider>();
_runInMemory = BusConfigurator is null;

9
src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs
View file

@ -4,8 +4,6 @@ using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.MassTransit.ConsumerDefinitions;
using Elsa.MassTransit.Consumers;
using Elsa.MassTransit.Contracts;
using Elsa.MassTransit.Formatters;
using Elsa.MassTransit.Options;
using Elsa.MassTransit.Services;
using Elsa.Workflows.Runtime.Contracts;
@ -31,11 +29,7 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
/// Configures the MassTransit workflow dispatcher.
/// </summary>
public Action<MassTransitWorkflowDispatcherOptions>? ConfigureDispatcherOptions { get; set; }
/// <summary>
/// A factory that creates a <see cref="IEndpointChannelFormatter"/>.
/// </summary>
public Func<IServiceProvider, IEndpointChannelFormatter> ChannelQueueFormatterFactory { get; set; } = _ => new DefaultEndpointChannelFormatter();
/// <inheritdoc />
public override void Configure()
@ -62,7 +56,6 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
if (ConfigureDispatcherOptions != null)
options.Configure(ConfigureDispatcherOptions);
Services.AddSingleton(ChannelQueueFormatterFactory);
Services.AddScoped<MassTransitWorkflowCancellationDispatcher>();
}
}

78
src/modules/Elsa.MongoDb/Common/MongoDbStore.cs
View file

@ -15,18 +15,18 @@ namespace Elsa.MongoDb.Common;
public class MongoDbStore<TDocument> where TDocument : class
{
private readonly IMongoCollection<TDocument> _collection;
/// <param name="collection"></param>
public MongoDbStore(IMongoCollection<TDocument> collection)
{
_collection = collection;
}
/// <summary>
/// Returns a queryable collection of documents.
/// </summary>
public IMongoCollection<TDocument> GetCollection() => _collection;
/// <summary>
/// Saves the document.
/// </summary>
@ -37,7 +37,7 @@ public class MongoDbStore<TDocument> where TDocument : class
await _collection.InsertOneAsync(document, new InsertOneOptions(), cancellationToken);
return document;
}
/// <summary>
/// Saves a list of documents.
/// </summary>
@ -47,7 +47,7 @@ public class MongoDbStore<TDocument> where TDocument : class
{
await _collection.InsertManyAsync(documents, new InsertManyOptions(), cancellationToken);
}
/// <summary>
/// Saves the document.
/// </summary>
@ -55,7 +55,11 @@ public class MongoDbStore<TDocument> where TDocument : class
/// <param name="cancellationToken">The cancellation token.</param>
public async Task<TDocument> SaveAsync(TDocument document, CancellationToken cancellationToken = default)
{
return await _collection.FindOneAndReplaceAsync(document.BuildIdFilter(), document, new FindOneAndReplaceOptions<TDocument>{ ReturnDocument = ReturnDocument.After, IsUpsert = true }, cancellationToken);
return await _collection.FindOneAndReplaceAsync(document.BuildIdFilter(), document, new FindOneAndReplaceOptions<TDocument>
{
ReturnDocument = ReturnDocument.After,
IsUpsert = true
}, cancellationToken);
}
/// <summary>
@ -66,9 +70,13 @@ public class MongoDbStore<TDocument> where TDocument : class
/// <param name="cancellationToken">The cancellation token.</param>
public async Task<TDocument> SaveAsync<TResult>(TDocument document, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default)
{
return await _collection.FindOneAndReplaceAsync(document.BuildExpression(selector), document, new FindOneAndReplaceOptions<TDocument>{ ReturnDocument = ReturnDocument.After, IsUpsert = true }, cancellationToken);
return await _collection.FindOneAndReplaceAsync(document.BuildExpression(selector), document, new FindOneAndReplaceOptions<TDocument>
{
ReturnDocument = ReturnDocument.After,
IsUpsert = true
}, cancellationToken);
}
/// <summary>
/// Saves the specified documents.
/// </summary>
@ -80,10 +88,16 @@ public class MongoDbStore<TDocument> where TDocument : class
foreach (var document in documents)
{
var replacement = new ReplaceOneModel<TDocument>(document.BuildIdFilter(), document) { IsUpsert = true };
var replacement = new ReplaceOneModel<TDocument>(document.BuildIdFilter(), document)
{
IsUpsert = true
};
writes.Add(replacement);
}
if (!writes.Any())
return;
await _collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
}
@ -99,10 +113,16 @@ public class MongoDbStore<TDocument> where TDocument : class
foreach (var document in documents)
{
var replacement = new ReplaceOneModel<TDocument>(document.BuildFilter(primaryKey), document) { IsUpsert = true };
var replacement = new ReplaceOneModel<TDocument>(document.BuildFilter(primaryKey), document)
{
IsUpsert = true
};
writes.Add(replacement);
}
if (!writes.Any())
return;
await _collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
}
@ -112,60 +132,60 @@ public class MongoDbStore<TDocument> where TDocument : class
/// <param name="predicate">The predicate to use.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The document if found, otherwise <c>null</c>.</returns>
public async Task<TDocument?> FindAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
public async Task<TDocument?> FindAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
await _collection.AsQueryable().Where(predicate).FirstOrDefaultAsync(cancellationToken);
/// <summary>
/// Finds a single document using a query
/// </summary>
/// <param name="query">The query to use</param>
/// <param name="cancellationToken">The cancellation token</param>
/// <returns>The document if found, otherwise <c>null</c></returns>
public async Task<TDocument?> FindAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
public async Task<TDocument?> FindAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
await query(_collection.AsQueryable()).FirstOrDefaultAsync(cancellationToken);
/// <summary>
/// Finds a list of documents matching the specified predicate
/// </summary>
public async Task<IEnumerable<TDocument>> FindManyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
public async Task<IEnumerable<TDocument>> FindManyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
await _collection.AsQueryable().Where(predicate).ToListAsync(cancellationToken);
/// <summary>
/// Queries the database using a query and a selector.
/// </summary>
public async Task<IEnumerable<TResult>> FindManyAsync<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) =>
public async Task<IEnumerable<TResult>> FindManyAsync<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) =>
await query(_collection.AsQueryable()).Select(selector).ToListAsync(cancellationToken);
/// <summary>
/// Finds a list of documents using a query
/// </summary>
public async Task<IEnumerable<TDocument>> FindManyAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
public async Task<IEnumerable<TDocument>> FindManyAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
await query(_collection.AsQueryable()).ToListAsync(cancellationToken);
/// <summary>
/// Queries the database using a query and a selector.
/// </summary>
public async Task<IEnumerable<TResult>> FindMany<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) =>
public async Task<IEnumerable<TResult>> FindMany<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) =>
await query(_collection.AsQueryable()).Select(selector).ToListAsync(cancellationToken);
/// <summary>
/// Counts documents in the collection using a filter.
/// </summary>
public async Task<long> CountAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
public async Task<long> CountAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) =>
await query(_collection.AsQueryable()).LongCountAsync(cancellationToken);
/// <summary>
/// Counts documents in the collection using a filter and distinct by a key selector.
/// </summary>
public async Task<long> CountAsync<TProperty>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TProperty>> propertySelector, CancellationToken cancellationToken = default) =>
public async Task<long> CountAsync<TProperty>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TProperty>> propertySelector, CancellationToken cancellationToken = default) =>
await query((IMongoQueryable<TDocument>)_collection.AsQueryable().DistinctBy(propertySelector)).LongCountAsync(cancellationToken);
/// <summary>
/// Checks if any documents exist.
/// </summary>
public async Task<bool> AnyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
public async Task<bool> AnyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) =>
await _collection.AsQueryable().Where(predicate).AnyAsync(cancellationToken);
/// <summary>
/// Deletes documents using a predicate.
/// </summary>
@ -174,12 +194,12 @@ public class MongoDbStore<TDocument> where TDocument : class
{
var documentsToDelete = await _collection.AsQueryable().Where(predicate).ToListAsync(cancellationToken);
var count = documentsToDelete.LongCount();
await _collection.DeleteManyAsync(predicate, cancellationToken);
return count;
}
/// <summary>
/// Deletes documents using a query.
/// </summary>
@ -189,7 +209,7 @@ public class MongoDbStore<TDocument> where TDocument : class
var key = keySelector.GetPropertyName();
return await DeleteWhereAsync(query, key, cancellationToken);
}
/// <summary>
/// Deletes documents using a query.
/// </summary>

6
src/modules/Elsa.Workflows.Api/Features/WorkflowsApiFeature.cs
View file

@ -1,4 +1,3 @@
using Elsa.Common.Contracts;
using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
@ -6,10 +5,8 @@ using Elsa.Features.Services;
using Elsa.Http.Features;
using Elsa.SasTokens.Features;
using Elsa.Workflows.Api.Serialization;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Management.Features;
using Elsa.Workflows.Services;
using Microsoft.Extensions.DependencyInjection;
using Elsa.Workflows.Runtime.Features;
namespace Elsa.Workflows.Api.Features;
@ -18,6 +15,7 @@ namespace Elsa.Workflows.Api.Features;
/// </summary>
[DependsOn(typeof(WorkflowInstancesFeature))]
[DependsOn(typeof(WorkflowManagementFeature))]
[DependsOn(typeof(WorkflowRuntimeFeature))]
[DependsOn(typeof(HttpFeature))]
[DependsOn(typeof(SasTokensFeature))]
public class WorkflowsApiFeature : FeatureBase

2
src/modules/Elsa.Workflows.Core/Models/DynamicActivity.cs → src/modules/Elsa.Workflows.Core/Activities/DynamicActivity.cs
View file

@ -1,7 +1,7 @@
using System.ComponentModel;
using Elsa.Workflows.Services;
namespace Elsa.Workflows.Models;
namespace Elsa.Workflows.Activities;
/// <summary>
/// A dynamically provided activity with custom properties. This is experimental and may be removed.

18
src/modules/Elsa.Workflows.Core/Contexts/ActivityCompletedContext.cs
View file

@ -1,3 +1,6 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Workflows.Activities.Flowchart.Models;
namespace Elsa.Workflows;
/// <summary>
@ -21,4 +24,19 @@ public record ActivityCompletedContext(ActivityExecutionContext TargetContext, A
/// A cancellation token to use when invoking asynchronous operations.
/// </summary>
public CancellationToken CancellationToken => WorkflowExecutionContext.CancellationTokens.ApplicationCancellationToken;
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public async ValueTask CompleteActivityAsync(object? result = default)
{
await TargetContext.CompleteActivityAsync(result);
}
/// <summary>
/// Complete the current activity with the specified outcomes.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes) => CompleteActivityAsync(new Outcomes(outcomes));
}

8
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Cancel.cs
View file

@ -19,9 +19,17 @@ public partial class ActivityExecutionContext
_ = Task.Run(async () => await CancelActivityAsync());
}
private bool CanCancelActivity()
{
return Status is not ActivityStatus.Canceled and not ActivityStatus.Completed;
}
private async Task CancelActivityAsync()
{
if(!CanCancelActivity())
return;
// Select all child contexts.
var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this).ToList();

109
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs Normal file
View file

@ -0,0 +1,109 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Extensions;
using Elsa.Workflows.Activities.Flowchart.Models;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Signals;
namespace Elsa.Workflows;
public partial class ActivityExecutionContext
{
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public async ValueTask CompleteActivityAsync(object? result = default)
{
var outcomes = result as Outcomes;
// If the activity is executing in the background, simply capture the result and return.
if (this.GetIsBackgroundExecution())
{
if (outcomes != null)
this.SetBackgroundOutcomes(outcomes.Names);
else
this.SetBackgroundCompletion();
return;
}
// If the activity is not running, do nothing.
if (Status != ActivityStatus.Running)
return;
// Cancel any non-completed child activities.
var childContexts = WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == this && x.CanCancelActivity()).ToList();
foreach (var childContext in childContexts)
await childContext.CancelActivityAsync();
// Mark the activity as complete.
TransitionTo(ActivityStatus.Completed);
// Record the outcomes, if any.
if (outcomes != null)
JournalData["Outcomes"] = outcomes.Names;
// Record the output, if any.
var activity = Activity;
var expressionExecutionContext = ExpressionExecutionContext;
var activityDescriptor = ActivityDescriptor;
var outputDescriptors = activityDescriptor.Outputs;
var outputs = outputDescriptors.ToDictionary(x => x.Name, x => activity.GetOutput(expressionExecutionContext, x.Name)!);
var serializer = GetRequiredService<ISafeSerializer>();
foreach (var outputDescriptor in outputDescriptors)
{
if (outputDescriptor.IsSerializable == false)
continue;
var outputName = outputDescriptor.Name;
var outputValue = outputs[outputName];
if (outputValue == null!)
continue;
var serializedOutputValue = await serializer.SerializeAsync(outputValue, CancellationToken);
JournalData[outputName] = serializedOutputValue;
}
// Add an execution log entry.
AddExecutionLogEntry("Completed", payload: JournalData);
// Send a signal.
await this.SendSignalAsync(new ActivityCompleted(result));
// Clear bookmarks.
ClearBookmarks();
WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityInstanceId == Id);
// Remove completion callbacks.
ClearCompletionCallbacks();
// Remove all associated variables, unless this is the root context - in which case we want to keep the variables since we're not deleting that one.
if (ParentActivityExecutionContext != null)
{
var variablePersistenceManager = GetRequiredService<IVariablePersistenceManager>();
await variablePersistenceManager.DeleteVariablesAsync(this);
}
// Update the completed at timestamp.
CompletedAt = WorkflowExecutionContext.SystemClock.UtcNow;
}
/// <summary>
/// Complete the current activity with the specified outcomes.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public ValueTask CompleteActivityWithOutcomesAsync(params string[] outcomes)
{
return CompleteActivityAsync(new Outcomes(outcomes));
}
/// <summary>
/// Complete the current composite activity with the specified outcome.
/// </summary>
public async ValueTask CompleteCompositeAsync(params string[] outcomes)
{
await this.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes)));
}
}

28
src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs
View file

@ -136,13 +136,11 @@ public partial class ActivityExecutionContext : IExecutionContext
public void TransitionTo(ActivityStatus status)
{
Status = status;
if (Status is ActivityStatus.Completed
or ActivityStatus.Canceled
or ActivityStatus.Faulted)
if (Status is ActivityStatus.Completed or ActivityStatus.Canceled or ActivityStatus.Faulted)
_cancellationRegistration.Dispose();
}
/// <summary>
/// Gets or sets the exception that occurred during the activity execution, if any.
/// </summary>
@ -254,15 +252,17 @@ public partial class ActivityExecutionContext : IExecutionContext
{
ActivityNodeId = activityNode?.NodeId,
OwnerActivityInstanceId = owner?.Id,
Options = options != null ? new ScheduledActivityOptions
{
CompletionCallback = options?.CompletionCallback?.Method.Name,
Tag = options?.Tag,
ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id,
PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false,
Variables = options?.Variables?.ToList(),
Input = options?.Input
} : default
Options = options != null
? new ScheduledActivityOptions
{
CompletionCallback = options?.CompletionCallback?.Method.Name,
Tag = options?.Tag,
ExistingActivityInstanceId = options?.ExistingActivityExecutionContext?.Id,
PreventDuplicateScheduling = options?.PreventDuplicateScheduling ?? false,
Variables = options?.Variables?.ToList(),
Input = options?.Input
}
: default
};
var scheduledActivities = this.GetBackgroundScheduledActivities().ToList();

109
src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs
View file

@ -1,3 +1,4 @@
using System.Diagnostics.CodeAnalysis;
using System.Linq.Expressions;
using System.Reflection;
using System.Text.Json;
@ -6,7 +7,6 @@ using Elsa.Expressions.Helpers;
using Elsa.Expressions.Models;
using Elsa.Mediator.Contracts;
using Elsa.Workflows;
using Elsa.Workflows.Activities.Flowchart.Models;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Memory;
@ -406,95 +406,7 @@ public static class ActivityExecutionContextExtensions
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
public static async ValueTask CompleteActivityAsync(this ActivityCompletedContext context, object? result = default)
{
await context.TargetContext.CompleteActivityAsync(result);
}
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
public static async ValueTask CompleteActivityAsync(this ActivityExecutionContext context, object? result = default)
{
var outcomes = result as Outcomes;
// If the activity is executing in the background, simply capture the result and return.
if (context.GetIsBackgroundExecution())
{
if (outcomes != null)
context.SetBackgroundOutcomes(outcomes.Names);
else
context.SetBackgroundCompletion();
return;
}
// If the activity is not running, do nothing.
if (context.Status != ActivityStatus.Running)
return;
// Update all child contexts.
var childContexts = context.WorkflowExecutionContext.ActivityExecutionContexts.Where(x => x.ParentActivityExecutionContext == context).ToList();
foreach (var childContext in childContexts)
await childContext.CancelActivityAsync();
// Mark the activity as complete.
context.TransitionTo(ActivityStatus.Completed);
// Record the outcomes, if any.
if (outcomes != null)
context.JournalData["Outcomes"] = outcomes.Names;
// Record the output, if any.
var activity = context.Activity;
var expressionExecutionContext = context.ExpressionExecutionContext;
var activityDescriptor = context.ActivityDescriptor;
var outputDescriptors = activityDescriptor.Outputs;
var outputs = outputDescriptors.ToDictionary(x => x.Name, x => activity.GetOutput(expressionExecutionContext, x.Name)!);
var serializer = context.GetRequiredService<ISafeSerializer>();
foreach (var outputDescriptor in outputDescriptors)
{
if (outputDescriptor.IsSerializable == false)
continue;
var outputName = outputDescriptor.Name;
var outputValue = outputs[outputName];
if (outputValue == null!)
continue;
var serializedOutputValue = await serializer.SerializeAsync(outputValue);
context.JournalData[outputName] = serializedOutputValue;
}
// Add an execution log entry.
context.AddExecutionLogEntry("Completed", payload: context.JournalData);
// Send a signal.
await context.SendSignalAsync(new ActivityCompleted(result));
// Clear bookmarks.
context.ClearBookmarks();
context.WorkflowExecutionContext.Bookmarks.RemoveWhere(x => x.ActivityInstanceId == context.Id);
// Remove completion callbacks.
context.ClearCompletionCallbacks();
// Remove all associated variables, unless this is the root context - in which case we want to keep the variables since we're not deleting that one.
if (context.ParentActivityExecutionContext != null)
{
var variablePersistenceManager = context.GetRequiredService<IVariablePersistenceManager>();
await variablePersistenceManager.DeleteVariablesAsync(context);
}
// Update the completed at timestamp.
context.CompletedAt = context.WorkflowExecutionContext.SystemClock.UtcNow;
}
/// <summary>
/// Complete the current activity. This should only be called by activities that explicitly suppress automatic-completion.
/// </summary>
[RequiresUnreferencedCode("The activity may be serialized and executed in a different context.")]
public static async ValueTask ScheduleOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes)
{
var cancellationToken = context.CancellationToken;
@ -525,22 +437,7 @@ public static class ActivityExecutionContextExtensions
// Send a signal.
await context.SendSignalAsync(new ScheduleActivityOutcomes(outcomes));
}
/// <summary>
/// Complete the current activity with the specified outcome.
/// </summary>
public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityCompletedContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes));
/// <summary>
/// Complete the current activity with the specified outcome.
/// </summary>
public static ValueTask CompleteActivityWithOutcomesAsync(this ActivityExecutionContext context, params string[] outcomes) => context.CompleteActivityAsync(new Outcomes(outcomes));
/// <summary>
/// Complete the current composite activity with the specified outcome.
/// </summary>
public static async ValueTask CompleteCompositeAsync(this ActivityExecutionContext context, params string[] outcomes) => await context.SendSignalAsync(new CompleteCompositeSignal(new Outcomes(outcomes)));
/// <summary>
/// Cancel the activity. For blocking activities, it means their bookmarks will be removed. For job activities, the background work will be cancelled.
/// </summary>

15
src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs
View file

@ -137,6 +137,21 @@ public static class ActivityExtensions
var outputName = outputExpression.GetPropertyName();
return ((IActivity)activity!).GetOutput<T>(context, outputName);
}
/// <summary>
/// Gets the output with the specified name.
/// </summary>
/// <param name="activity">The activity.</param>
/// <param name="context">The context.</param>
/// <param name="outputExpression">The output expression.</param>
/// <typeparam name="TActivity">The type of the activity.</typeparam>
/// <typeparam name="T">The type of the output.</typeparam>
/// <returns>The output value.</returns>
public static T? GetOutput<TActivity, T>(this TActivity activity, ExpressionExecutionContext context, Expression<Func<TActivity, object?>> outputExpression)
{
var outputName = outputExpression.GetPropertyName();
return ((IActivity)activity!).GetOutput<T>(context, outputName);
}
/// <summary>
/// Gets the Result output of the specified activity.

6
src/modules/Elsa.Workflows.Core/Models/CreateBookmarkArgs.cs
View file

@ -16,14 +16,14 @@ public class CreateBookmarkArgs
/// <summary>An optional name to associate with the bookmark.</summary>
public string? BookmarkName { get; set; }
/// <summary>Whether or not the bookmark should be automatically burned when triggered.</summary>
/// <summary>Whether the bookmark should be automatically burned when triggered.</summary>
public bool AutoBurn { get; set; } = true;
/// <summary>Whether or not the activity instance ID should be included in the bookmark payload.</summary>
/// <summary>Whether the activity instance ID should be included in the bookmark payload.</summary>
public bool IncludeActivityInstanceId { get; set; }
/// <summary>
/// Whether or not the activity being resumed should be automatically completed if CallBack is not specified.
/// Whether the activity being resumed should be automatically completed if CallBack is not specified.
/// </summary>
public bool AutoComplete { get; set; } = true;

11
src/modules/Elsa.Workflows.Core/Services/BookmarkHasher.cs
View file

@ -1,3 +1,4 @@
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Elsa.Expressions.Contracts;
using Elsa.Workflows.Contracts;
@ -30,6 +31,7 @@ public class BookmarkHasher : IBookmarkHasher
}
/// <inheritdoc />
[RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize(Object, Type, JsonSerializerOptions)")]
public string Hash(string activityTypeName, object? payload, string? activityInstanceId = default)
{
var json = payload != null ? Serialize(payload) : null;
@ -47,5 +49,12 @@ public class BookmarkHasher : IBookmarkHasher
return hash;
}
private string Serialize(object payload) => JsonSerializer.Serialize(payload, payload.GetType(), _settings);
[RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize(Object, Type, JsonSerializerOptions)")]
private string Serialize(object payload)
{
if(payload is string s)
return s;
return JsonSerializer.Serialize(payload, payload.GetType(), _settings);
}
}

24
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs
View file

@ -189,13 +189,13 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = options.CorrelationId,
Input = options.Input,
Properties = options.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
InstanceId = options.WorkflowInstanceId,
CancellationTokens = options.CancellationTokens
InstanceId = options?.WorkflowInstanceId,
CancellationTokens = options?.CancellationTokens ?? default
};
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions);
@ -276,10 +276,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options)
{
var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options.ActivityInstanceId);
var correlationId = options.CorrelationId;
var workflowInstanceId = options.WorkflowInstanceId;
var activityInstanceId = options.ActivityInstanceId;
var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options?.ActivityInstanceId);
var correlationId = options?.CorrelationId;
var workflowInstanceId = options?.WorkflowInstanceId;
var activityInstanceId = options?.ActivityInstanceId;
var filter = new BookmarkFilter
{
Hash = hash,
@ -287,15 +287,15 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
WorkflowInstanceId = workflowInstanceId,
ActivityInstanceId = activityInstanceId
};
var bookmarks = await _bookmarkStore.FindManyAsync(filter, options.CancellationTokens.SystemCancellationToken);
var bookmarks = await _bookmarkStore.FindManyAsync(filter, options?.CancellationTokens.SystemCancellationToken ?? default);
return await ResumeWorkflowsAsync(
bookmarks,
new ResumeWorkflowRuntimeParams
{
CorrelationId = correlationId,
Input = options.Input,
CancellationTokens = options.CancellationTokens
Input = options?.Input,
CancellationTokens = options?.CancellationTokens ?? default
});
}