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

Add Elsa.Kafka Module for Kafka Integration with Message Sending and Receiving Activities (#6108)

Browse source 
* Add Kafka module with integration and example setup

Introduced the Kafka module providing consumer integration and activities into the project. This includes new classes for consumer handling, configuration, and activities. An example setup using Docker Compose is also added to facilitate development and testing.

* Refactor KafkaTransportMessage to inline Timestamp namespace

Simplify the namespace usage for the Timestamp type within the KafkaTransportMessage record. This change eliminates the need for a separate using directive for Timestamp, enhancing code readability and maintainability.

* Add support for handling Kafka transport messages

This commit introduces the capability to handle and trigger workflows based on Kafka transport messages. It adds a new handler, notifications, and updates the message stimulus to include correlating fields. Additionally, the Kafka consumers are now managed more modularly with updated startup tasks and mediator integration.

* Enable Kafka integration and fix Kafka options naming

Added support for Kafka integration in Elsa.Server.Web by setting up Kafka configurations in appsettings.json and updating Program.cs. Also, renamed `ConsumerConfigs` to `ConsumerDefinitions` in Kafka options for clarity.

* Add consumer definition enumeration and dropdown support

Introduced `IConsumerDefinitionEnumerator` for managing consumer definitions across providers and implemented in `ConsumerDefinitionEnumerator` class. Enhanced `KafkaFeature` to register these services and updated the `MessageReceived` activity to use a dropdown UI hint for consuming definitions. Improved `StartConsumersTask` by refactoring consumer definition retrieval logic.

* Add SendMessage activity and refine Kafka messaging

Introduce a new SendMessage activity for Kafka, enabling message publishing to specific topics. Refine KafkaTransportMessage model by removing headers and timestamp fields. Adjust the StimulusSender logic to streamline the bookmark queuing process and fix key-value pairing in dropdown options. Update appsettings for corrected Kafka bootstrap server and topic configurations.

* Add producer and topic management support

Introduced interfaces and implementations for managing producer and topic definitions along with their respective enumerators and list providers. Updated `SendMessage` activity to include producer selection and refactored consumer definition providers for better consistency.

* Refactor Kafka configuration property names

Renamed Kafka configuration properties for better consistency and readability across the codebase. Updated property names from `ProducerDefinitions` to `Producers`, `ConsumerDefinitions` to `Consumers`, and `TopicDefinitions` to `Topics`. Added missing input attribute in `SendMessage.cs` and registered additional handlers in `KafkaFeature.cs`.

* Add custom serializers for Kafka message handling

Introduced `DefaultSerializers` class for custom serialization and deserialization of Kafka messages. Updated `MessageReceived` and `SendMessage` activities to use these custom serializers, and modified `KafkaOptions` to include them.

* Fix ExpandoObject serialization method parameter

Changed the serialization type from `ExpandoObject` to the actual type of the object to ensure proper serialization. This ensures that derived types are correctly handled during the serialization process.

* Add Producer and Consumer workflows for Kafka

Introduced two new workflows: `ProducerWorkflow` and `ConsumerWorkflow` for handling Kafka messages. Updated `DefaultSerializers` to use camelCase property naming and modified `appsettings.json` to include `topic-2` and format entries.

* Add JSON serialization to log output in ConsumerWorkflow

This change enhances the log output by serializing messages to JSON format before writing them. The addition of System.Text.Json ensures that the message content is presented in a structured and standardized format in logs.

* Add correlation strategies and update Kafka features

Implemented HeaderCorrelationStrategy and NullCorrelationStrategy, and updated KafkaFeature to support customizable correlation strategies. Added correlation ID handling to Kafka transport messages and updated config and handlers accordingly.

* Add tenant accessor to ConsumerDefinitionWorkflowContextProvider

Integrated ITenantAccessor to the provider to support tenant-specific context loading. Updated the constructor and LoadAsync method to retrieve the tenant information and use it for context-specific operations.

* Switch to MySQL and disable Kafka

This commit changes the SQL database provider from SQLite to MySQL and disables Kafka use. It also includes necessary adjustments such as adding MySQL handling in configuration and connection setups, updating `docker-compose` to include MySQL services, and referencing MySQL projects in the `.csproj` file.

* Add UI property handlers to multiple features

This commit introduces various UI property handlers across several features such as Python, JavaScript, CSharp, and Workflow features to enhance user interface property handling. It also updates the property UI handler resolution logic to better manage cases where providers are not available. Furthermore, adjustments were made in the server configuration to switch database providers and enable Kafka.

* Refactor property UI handler retrieval logic

Modified the logic to fetch property UI handlers by preloading them into a list and then filtering. This change improves readability and potentially performance by reducing repetitive service provider calls.

* Incremental work on Kafka workers and predicate evaluation

* Merge BookmarkInvoker with BookmarkResumer

* Register IWorkerManager

* Change lifetime scope of WorkerManager to Singleton

* **Introduce topic subscription handling for Kafka workers**

Added `IWorkerTopicSubscriber` interface and its implementation for managing topic subscriptions. Enhanced workers to bind triggers and bookmarks dynamically based on existing data. Updated several classes and methods to support topic-based subscriptions and headers.

* Refactor trigger matching logic.

Extract trigger matching conditions into `IsMatchAsync` method for reuse. This enhances code maintainability and readability by reducing redundancy. The new `GetTopic` helper method isolates the topic retrieval logic.

* Add Name property to MassTransitActivityTypeProvider

This commit inserts the Name property in the returned object within the MassTransitActivityTypeProvider class. It ensures that the typeName is included, providing a clearer definition of the activity type.

* Add handling for deleted bookmarks and refactor bookmark removal

Added a new event handler for `BookmarksDeleted` to ensure removed bookmarks are processed correctly. Refactored the bookmark removal logic into a helper method to reduce code duplication and streamline the workflow.

* Switch to asynchronous bookmark queue processing

Refactored the `TriggerWorkflows` handler to use `IBookmarkQueue` instead of directly invoking the `IBookmarkResumer`. This change aims to improve scalability by queueing bookmark resumption requests, enabling better load distribution and async processing. Added necessary helpers and configuration options to support this functionality.

* Remove unused IBookmarkResumer dependency

Simplify the constructor by removing the unused IBookmarkResumer dependency. This cleanup reduces potential confusion and improves code maintainability without impacting functionality.

* Add support for local message processing

Introduced an `IsLocal` property to `MessageReceivedStimulus` for determining if the message event is local to a specific workflow instance. Updated `BookmarkBinding` and related handler methods to utilize `CorrelationId` for local event matching. Removed unused `CorrelatingFields` from `MessageReceived` activity.

* Add nullability checks to IWorker retrieval methods

Updated `GetWorker` methods to return nullable `IWorker` to handle cases where a worker might not exist. Modified code to include null checks and conditional operations to prevent potential null reference exceptions when accessing worker methods.

* Add filtering based on activity type name for triggers and bookmarks

This commit introduces filtering for triggers and bookmarks based on the `MessageReceived` activity type name. It also adds an option to mark messages as local in the `SendMessage` activity, where local messages are delivered to the current workflow instance only. These changes help enhance the management and targeted delivery of messages within the workflow framework.

* Add new Kafka topics and clean up producers config

New topics "topic-3" and "topic-4" were added to the Kafka settings. Unused topic references were removed from the producers configuration to simplify and improve clarity.

* Add predicate to KafkaConsumerActivity in ConsumerWorkflow

Introduced a predicate to the KafkaConsumerActivity using JavaScript expressions to filter messages based on OrderId. This ensures only relevant messages are processed in the workflow.
This commit is contained in:
Sipke Schoorstra 2024-11-18 13:42:54 +01:00 committed by GitHub
parent 153397e63c
commit f53e024d25
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
78 changed files with 2006 additions and 122 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Directory.Packages.props
View file

@ -15,6 +15,7 @@
<PackageVersion Include="BenchmarkDotNet" Version="0.14.0" />
<PackageVersion Include="Bogus" Version="35.6.1" />
<PackageVersion Include="ConfigureAwait.Fody" Version="3.3.2" PrivateAssets="All" />
<PackageVersion Include="Confluent.Kafka" Version="2.6.0" />
<PackageVersion Include="coverlet.collector" Version="6.0.2" PrivateAssets="All" />
<PackageVersion Include="Cronos" Version="0.8.4" />
<PackageVersion Include="Dapper" Version="2.1.35" />
@ -84,7 +85,7 @@
<PackageVersion Include="pythonnet" Version="3.0.4" />
<PackageVersion Include="Quartz.Extensions.Hosting" Version="3.13.0" />
<PackageVersion Include="Quartz.Serialization.Json" Version="3.13.0" />
<PackageVersion Include="Scrutor" Version="5.0.2"/>
<PackageVersion Include="Scrutor" Version="5.0.2" />
<PackageVersion Include="ShortGuid" Version="2.0.1" />
<PackageVersion Include="StackExchange.Redis" Version="2.8.16" />
<PackageVersion Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />

8
Elsa.sln
View file

@ -91,6 +91,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "docker", "docker", "{986E54
docker\ElsaStudio.Dockerfile = docker\ElsaStudio.Dockerfile
docker\init-db.sh = docker\init-db.sh
docker\otel-collector-config.yaml = docker\otel-collector-config.yaml
docker\docker-compose-kafka.yml = docker\docker-compose-kafka.yml
EndProjectSection
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Elasticsearch", "src\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj", "{3246883E-2FA7-4B4A-BDC5-99039A2869BC}"
@ -384,6 +385,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Tenants.AspNetCore", "
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql", "src\modules\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql.csproj", "{2B939AC9-03A4-479E-AA0D-CB58F4A7F480}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Kafka", "src\modules\Elsa.Kafka\Elsa.Kafka.csproj", "{BF934627-F531-44FB-BEC2-ECA801FF31E7}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -816,6 +819,10 @@ Global
{2B939AC9-03A4-479E-AA0D-CB58F4A7F480}.Debug|Any CPU.Build.0 = Debug|Any CPU
{2B939AC9-03A4-479E-AA0D-CB58F4A7F480}.Release|Any CPU.ActiveCfg = Release|Any CPU
{2B939AC9-03A4-479E-AA0D-CB58F4A7F480}.Release|Any CPU.Build.0 = Release|Any CPU
{BF934627-F531-44FB-BEC2-ECA801FF31E7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{BF934627-F531-44FB-BEC2-ECA801FF31E7}.Debug|Any CPU.Build.0 = Debug|Any CPU
{BF934627-F531-44FB-BEC2-ECA801FF31E7}.Release|Any CPU.ActiveCfg = Release|Any CPU
{BF934627-F531-44FB-BEC2-ECA801FF31E7}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -960,6 +967,7 @@ Global
{D5720DBC-8C2B-42D5-9D9F-2FF6EAD4001C} = {2F3E1026-5054-4E1F-899B-F1A7F70F9912}
{2B939AC9-03A4-479E-AA0D-CB58F4A7F480} = {50470834-4CD8-479A-8B58-0A1869BA5D37}
{2CDF3E1C-267D-4198-B1C7-7E1F548FC120} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
{BF934627-F531-44FB-BEC2-ECA801FF31E7} = {DD089B8B-DA73-492A-9010-F772D1C178DA}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E}

167
docker/docker-compose-kafka.yml Normal file
View file

@ -0,0 +1,167 @@
---
version: '2'
services:
broker:
image: confluentinc/cp-kafka:7.7.1
hostname: broker
container_name: broker
ports:
- "9092:9092"
- "9101:9101"
environment:
KAFKA_NODE_ID: 1
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_JMX_PORT: 9101
KAFKA_JMX_HOSTNAME: localhost
KAFKA_PROCESS_ROLES: 'broker,controller'
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'
KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
# Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
# See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
schema-registry:
image: confluentinc/cp-schema-registry:7.7.1
hostname: schema-registry
container_name: schema-registry
depends_on:
- broker
ports:
- "8081:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
connect:
image: cnfldemos/cp-server-connect-datagen:0.6.4-7.6.0
hostname: connect
container_name: connect
depends_on:
- broker
- schema-registry
ports:
- "8083:8083"
environment:
CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'
CONNECT_REST_ADVERTISED_HOST_NAME: connect
CONNECT_GROUP_ID: compose-connect-group
CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
# CLASSPATH required due to CC-2422
CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.7.1.jar
CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR
control-center:
image: confluentinc/cp-enterprise-control-center:7.7.1
hostname: control-center
container_name: control-center
depends_on:
- broker
- schema-registry
- connect
- ksqldb-server
ports:
- "9021:9021"
environment:
CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
CONTROL_CENTER_CONNECT_HEALTHCHECK_ENDPOINT: '/connectors'
CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
CONTROL_CENTER_REPLICATION_FACTOR: 1
CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
CONFLUENT_METRICS_TOPIC_REPLICATION: 1
PORT: 9021
ksqldb-server:
image: confluentinc/cp-ksqldb-server:7.7.1
hostname: ksqldb-server
container_name: ksqldb-server
depends_on:
- broker
- connect
ports:
- "8088:8088"
environment:
KSQL_CONFIG_DIR: "/etc/ksql"
KSQL_BOOTSTRAP_SERVERS: "broker:29092"
KSQL_HOST_NAME: ksqldb-server
KSQL_LISTENERS: "http://0.0.0.0:8088"
KSQL_CACHE_MAX_BYTES_BUFFERING: 0
KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
KSQL_KSQL_CONNECT_URL: "http://connect:8083"
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_REPLICATION_FACTOR: 1
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: 'true'
KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: 'true'
ksqldb-cli:
image: confluentinc/cp-ksqldb-cli:7.7.1
container_name: ksqldb-cli
depends_on:
- broker
- connect
- ksqldb-server
entrypoint: /bin/sh
tty: true
ksql-datagen:
image: confluentinc/ksqldb-examples:7.7.1
hostname: ksql-datagen
container_name: ksql-datagen
depends_on:
- ksqldb-server
- broker
- schema-registry
- connect
command: "bash -c 'echo Waiting for Kafka to be ready... && \
cub kafka-ready -b broker:29092 1 40 && \
echo Waiting for Confluent Schema Registry to be ready... && \
cub sr-ready schema-registry 8081 40 && \
echo Waiting a few seconds for topic creation to finish... && \
sleep 11 && \
tail -f /dev/null'"
environment:
KSQL_CONFIG_DIR: "/etc/ksql"
STREAMS_BOOTSTRAP_SERVERS: broker:29092
STREAMS_SCHEMA_REGISTRY_HOST: schema-registry
STREAMS_SCHEMA_REGISTRY_PORT: 8081
rest-proxy:
image: confluentinc/cp-kafka-rest:7.7.1
depends_on:
- broker
- schema-registry
ports:
- 8082:8082
hostname: rest-proxy
container_name: rest-proxy
environment:
KAFKA_REST_HOST_NAME: rest-proxy
KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'

14
docker/docker-compose.yml
View file

@ -11,6 +11,19 @@
- ./init-db.sh:/docker-entrypoint-initdb.d/init-db.sh # This will initialize the 'tracelens' database
ports:
- "5432:5432"
mysql:
image: mysql:9.1.0
container_name: mysql
environment:
MYSQL_ROOT_PASSWORD: password
MYSQL_DATABASE: elsa
MYSQL_USER: admin
MYSQL_PASSWORD: password
ports:
- "3306:3306"
volumes:
- mysql_data:/var/lib/mysql
mongodb:
image: mongo:latest
@ -91,5 +104,6 @@
volumes:
postgres-data:
mysql_data:
cockroachdb-data:
mongodb_data:

4
src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj
View file

@ -3,13 +3,17 @@
<ItemGroup>
<ProjectReference Include="..\..\modules\Elsa.Agents.Activities\Elsa.Agents.Activities.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Agents.Api\Elsa.Agents.Api.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.MySql\Elsa.Agents.Persistence.EntityFrameworkCore.MySql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.Sqlite\Elsa.Agents.Persistence.EntityFrameworkCore.Sqlite.csproj" />
<ProjectReference Include="..\..\modules\Elsa.AzureServiceBus\Elsa.AzureServiceBus.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.ProtoActor\Elsa.Caching.Distributed.ProtoActor.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.MySql\Elsa.EntityFrameworkCore.MySql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.Kafka\Elsa.Kafka.csproj" />
<ProjectReference Include="..\..\modules\Elsa.MassTransit.AzureServiceBus\Elsa.MassTransit.AzureServiceBus.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.OpenTelemetry\Elsa.OpenTelemetry.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Quartz.EntityFrameworkCore.MySql\Elsa.Quartz.EntityFrameworkCore.MySql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Api\Elsa.Secrets.Api.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Secrets.Persistence.EntityFrameworkCore.PostgreSql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Persistence.EntityFrameworkCore.Sqlite\Elsa.Secrets.Persistence.EntityFrameworkCore.Sqlite.csproj" />

22
src/apps/Elsa.Server.Web/Program.cs
View file

@ -16,6 +16,7 @@ using Elsa.EntityFrameworkCore.Modules.Runtime;
using Elsa.Extensions;
using Elsa.Features.Services;
using Elsa.Identity.Multitenancy;
using Elsa.Kafka;
using Elsa.MassTransit.Extensions;
using Elsa.MongoDb.Extensions;
using Elsa.MongoDb.Modules.Alterations;
@ -30,6 +31,7 @@ using Elsa.Server.Web;
using Elsa.Server.Web.Extensions;
using Elsa.Server.Web.Filters;
using Elsa.Server.Web.Messages;
using Elsa.Server.Web.WorkflowContextProviders;
using Elsa.Tenants.AspNetCore;
using Elsa.Tenants.Extensions;
using Elsa.Workflows.Api;
@ -63,6 +65,7 @@ const bool runEFCoreMigrations = true;
const bool useMemoryStores = false;
const bool useCaching = true;
const bool useAzureServiceBus = false;
const bool useKafka = true;
const bool useReadOnlyMode = false;
const bool useSignalR = false; // Disabled until Elsa Studio sends authenticated requests.
const WorkflowRuntime workflowRuntime = WorkflowRuntime.ProtoActor;
@ -81,6 +84,7 @@ var identityTokenSection = identitySection.GetSection("Tokens");
var sqliteConnectionString = configuration.GetConnectionString("Sqlite")!;
var sqlServerConnectionString = configuration.GetConnectionString("SqlServer")!;
var postgresConnectionString = configuration.GetConnectionString("PostgreSql")!;
var mySqlConnectionString = configuration.GetConnectionString("MySql")!;
var cockroachDbConnectionString = configuration.GetConnectionString("CockroachDb")!;
var mongoDbConnectionString = configuration.GetConnectionString("MongoDb")!;
var azureServiceBusConnectionString = configuration.GetConnectionString("AzureServiceBus")!;
@ -136,6 +140,8 @@ services
ef.UseSqlServer(sqlServerConnectionString!);
else if (sqlDatabaseProvider == SqlDatabaseProvider.PostgreSql)
ef.UsePostgreSql(postgresConnectionString!);
else if(sqlDatabaseProvider == SqlDatabaseProvider.MySql)
ef.UseMySql(mySqlConnectionString);
else if (sqlDatabaseProvider == SqlDatabaseProvider.CockroachDb)
ef.UsePostgreSql(cockroachDbConnectionString!);
else
@ -168,6 +174,8 @@ services
ef.UseSqlServer(sqlServerConnectionString!);
else if (sqlDatabaseProvider == SqlDatabaseProvider.PostgreSql)
ef.UsePostgreSql(postgresConnectionString!);
else if(sqlDatabaseProvider == SqlDatabaseProvider.MySql)
ef.UseMySql(mySqlConnectionString);
else if (sqlDatabaseProvider == SqlDatabaseProvider.CockroachDb)
ef.UsePostgreSql(cockroachDbConnectionString!);
else
@ -236,6 +244,8 @@ services
}
else if (sqlDatabaseProvider == SqlDatabaseProvider.PostgreSql)
ef.UsePostgreSql(postgresConnectionString!);
else if(sqlDatabaseProvider == SqlDatabaseProvider.MySql)
ef.UseMySql(mySqlConnectionString);
else if (sqlDatabaseProvider == SqlDatabaseProvider.CockroachDb)
ef.UsePostgreSql(cockroachDbConnectionString!);
else
@ -359,6 +369,8 @@ services
ef.UseSqlServer(sqlServerConnectionString);
else if (sqlDatabaseProvider == SqlDatabaseProvider.PostgreSql)
ef.UsePostgreSql(postgresConnectionString);
else if(sqlDatabaseProvider == SqlDatabaseProvider.MySql)
ef.UseMySql(mySqlConnectionString);
else if (sqlDatabaseProvider == SqlDatabaseProvider.CockroachDb)
ef.UsePostgreSql(cockroachDbConnectionString!);
else
@ -435,6 +447,16 @@ services
asb.AzureServiceBusOptions = options => configuration.GetSection("AzureServiceBus").Bind(options);
});
}
if(useKafka)
{
elsa.UseKafka(kafka =>
{
kafka.ConfigureOptions(options => configuration.GetSection("Kafka").Bind(options));
});
services.AddWorkflowContextProvider<ConsumerDefinitionWorkflowContextProvider>();
}
if (useAgents)
{

22
src/apps/Elsa.Server.Web/WorkflowContextProviders/ConsumerConfigWorkflowContextProvider.cs Normal file
View file

@ -0,0 +1,22 @@
using Elsa.Common.Multitenancy;
using Elsa.Kafka;
using Elsa.WorkflowContexts.Abstractions;
using Elsa.Workflows;
namespace Elsa.Server.Web.WorkflowContextProviders;
public class ConsumerDefinitionWorkflowContextProvider(IConsumerDefinitionEnumerator consumerDefinitionEnumerator, ITenantAccessor tenantAccessor) : WorkflowContextProvider<ConsumerDefinition>
{
protected override async ValueTask<ConsumerDefinition?> LoadAsync(WorkflowExecutionContext workflowExecutionContext)
{
var tenant = tenantAccessor.Tenant;
var tenantId = tenant?.Id;
var definitionId = workflowExecutionContext.Workflow.Identity.DefinitionId;
// Load specific setting here.
// For now, just return the first consumer definition.
var consumerDefinitions = await consumerDefinitionEnumerator.EnumerateAsync(workflowExecutionContext.CancellationToken);
return consumerDefinitions.FirstOrDefault();
}
}

33
src/apps/Elsa.Server.Web/Workflows/ConsumerWorkflow.cs Normal file
View file

@ -0,0 +1,33 @@
using System.Dynamic;
using System.Text.Json;
using Elsa.JavaScript.Models;
using Elsa.Kafka.Activities;
using Elsa.Workflows;
using Elsa.Workflows.Activities;
namespace Elsa.Server.Web.Workflows;
public class ConsumerWorkflow : WorkflowBase
{
protected override void Build(IWorkflowBuilder builder)
{
var message = builder.WithVariable<ExpandoObject>();
builder.Name = "Consumer Workflow";
builder.Root = new Sequence
{
Activities =
{
new MessageReceived
{
ConsumerDefinitionId = new("consumer-1"),
MessageType = new(typeof(ExpandoObject)),
Topics = new(["topic-1"]),
Predicate = new(JavaScriptExpression.Create("message => message.OrderId == '1'")),
Result = new(message),
CanStartWorkflow = true
},
new WriteLine(c => JsonSerializer.Serialize(message.Get(c)))
}
};
}
}

22
src/apps/Elsa.Server.Web/Workflows/ProducerWorkflow.cs Normal file
View file

@ -0,0 +1,22 @@
using Elsa.Kafka.Activities;
using Elsa.Workflows;
namespace Elsa.Server.Web.Workflows;
public class ProducerWorkflow : WorkflowBase
{
protected override void Build(IWorkflowBuilder builder)
{
builder.Name = "Producer Workflow";
builder.Root = new SendMessage
{
Topic = new("topic-2"),
ProducerDefinitionId = new("producer-1"),
Content = new(() => new
{
OrderId = "1",
CustomerId = "1"
})
};
}
}

56
src/apps/Elsa.Server.Web/appsettings.json
View file

@ -9,6 +9,7 @@
"ConnectionStrings": {
"Sqlite": "Data Source=App_Data/elsa.sqlite.db;Cache=Shared;",
"PostgreSql": "Server=localhost;Username=elsa;Database=elsa;Port=5432;Password=elsa;SSLMode=Prefer;MaxPoolSize=2000;Timeout=60",
"MySql": "Server=localhost;Port=3306;Database=elsa;User=admin;Password=password;",
"CockroachDb": "Host=localhost;Port=26257;Database=elsa;SslMode=Disable;Username=root;IncludeErrorDetail=true",
"MongoDb": "mongodb://localhost:27017/elsa-workflows",
"AzureServiceBus": "",
@ -196,6 +197,61 @@
}
]
},
"Kafka": {
"Topics": [
{
"Id": "topic-1",
"Name": "topic-1"
},
{
"Id": "topic-2",
"Name": "topic-2"
},
{
"Id": "topic-3",
"Name": "topic-3"
},
{
"Id": "topic-4",
"Name": "topic-4"
}
],
"Producers": [
{
"Id": "producer-1",
"Name": "Producer 1",
"BootstrapServers": [
"localhost:9092"
]
}
],
"Consumers": [
{
"Id": "consumer-1",
"Name": "Consumer 1",
"BootstrapServers": [
"localhost:9092"
],
"GroupId": "group-1",
"AutoOffsetReset": "earliest",
"EnableAutoCommit": "true",
"CorrelatingFields": [
"orderId",
"customerId"
]
},
{
"Id": "consumer-2",
"Name": "Consumer 2",
"BootstrapServers": [
"localhost:9092"
],
"GroupId": "group-1",
"AutoOffsetReset": "earliest",
"EnableAutoCommit": "true"
}
]
},
"Secrets": {
"Management": {
"SweepInterval": "04:00:00"

5
src/modules/Elsa.CSharp/Features/CSharpFeature.cs
View file

@ -1,4 +1,5 @@
using Elsa.Common.Features;
using Elsa.CSharp.Activities;
using Elsa.CSharp.Contracts;
using Elsa.CSharp.Options;
using Elsa.CSharp.Providers;
@ -8,6 +9,7 @@ using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.CSharp.Features;
@ -45,5 +47,8 @@ public class CSharpFeature : FeatureBase
// Activities.
Module.AddActivitiesFrom<CSharpFeature>();
// UI property handlers.
Services.AddScoped<IPropertyUIHandler, RunCSharpOptionsProvider>();
}
}

34
src/modules/Elsa.Expressions/Models/ExpressionExecutionContext.cs
View file

@ -6,49 +6,37 @@ namespace Elsa.Expressions.Models;
/// <summary>
/// Provides context to workflow expressions.
/// </summary>
public class ExpressionExecutionContext
public class ExpressionExecutionContext(
IServiceProvider serviceProvider,
MemoryRegister memory,
ExpressionExecutionContext? parentContext = default,
IDictionary<object, object>? transientProperties = default,
CancellationToken cancellationToken = default)
{
/// <summary>
/// Constructor.
/// </summary>
public ExpressionExecutionContext(
IServiceProvider serviceProvider,
MemoryRegister memory,
ExpressionExecutionContext? parentContext = default,
IDictionary<object, object>? transientProperties = default,
CancellationToken cancellationToken = default)
{
ServiceProvider = serviceProvider;
Memory = memory;
TransientProperties = transientProperties ?? new Dictionary<object, object>();
ParentContext = parentContext;
CancellationToken = cancellationToken;
}
/// <summary>
/// A scoped service provider.
/// </summary>
public IServiceProvider ServiceProvider { get; }
public IServiceProvider ServiceProvider { get; } = serviceProvider;
/// <summary>
/// A shared register of computer memory.
/// </summary>
public MemoryRegister Memory { get; }
public MemoryRegister Memory { get; } = memory;
/// <summary>
/// A dictionary of transient properties.
/// </summary>
public IDictionary<object, object> TransientProperties { get; set; }
public IDictionary<object, object> TransientProperties { get; set; } = transientProperties ?? new Dictionary<object, object>();
/// <summary>
/// Provides access to the parent <see cref="ExpressionExecutionContext"/>, if there is any.
/// </summary>
public ExpressionExecutionContext? ParentContext { get; set; }
public ExpressionExecutionContext? ParentContext { get; set; } = parentContext;
/// <summary>
/// A cancellation token.
/// </summary>
public CancellationToken CancellationToken { get; }
public CancellationToken CancellationToken { get; } = cancellationToken;
/// <summary>
/// Returns the <see cref="MemoryBlock"/> pointed to by the specified memory block reference.

1
src/modules/Elsa.Http/Features/HttpFeature.cs
View file

@ -186,6 +186,7 @@ public class HttpFeature(IModule module) : FeatureBase(module)
// Activity property options providers.
.AddScoped<IPropertyUIHandler, HttpContentTypeOptionsProvider>()
.AddScoped<IPropertyUIHandler, HttpEndpointPathUIHandler>()
.AddScoped(_httpEndpointBasePathProvider)
// Port resolvers.

10
src/modules/Elsa.JavaScript/Features/JavaScriptFeature.cs
View file

@ -36,7 +36,7 @@ public class JavaScriptFeature : FeatureBase
/// Configures the Jint options.
/// </summary>
private Action<JintOptions> JintOptions { get; set; } = _ => { };
public JavaScriptFeature ConfigureJintOptions(Action<JintOptions> configure)
{
JintOptions += configure;
@ -88,8 +88,10 @@ public class JavaScriptFeature : FeatureBase
// Activities.
Module.UseWorkflowManagement(management => management.AddActivity<RunJavaScript>());
Services
.AddScoped<IPropertyUIHandler, RunJavaScriptOptionsProvider>()
.AddFunctionDefinitionProvider<InputFunctionsDefinitionProvider>();
// Type Script definitions.
Services.AddFunctionDefinitionProvider<InputFunctionsDefinitionProvider>();
// UI property handlers.
Services.AddScoped<IPropertyUIHandler, RunJavaScriptOptionsProvider>();
}
}

130
src/modules/Elsa.Kafka/Activities/MessageReceived.cs Normal file
View file

@ -0,0 +1,130 @@
using System.Runtime.CompilerServices;
using Elsa.Expressions.Models;
using Elsa.Extensions;
using Elsa.Kafka.Stimuli;
using Elsa.Kafka.UIHints;
using Elsa.Workflows;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Models;
using Elsa.Workflows.UIHints;
using Microsoft.Extensions.Options;
namespace Elsa.Kafka.Activities;
[Activity("Elsa.Kafka", "Kafka", "Executes when a message is received from a given set of topics")]
public class MessageReceived : Trigger<object>
{
internal const string InputKey = "TransportMessage";
/// <inheritdoc />
public MessageReceived([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
{
}
/// <inheritdoc />
public MessageReceived(Input<string> consumerDefinitionId, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
{
ConsumerDefinitionId = consumerDefinitionId;
}
/// <summary>
/// The consumer to read from.
/// </summary>
[Input(
DisplayName = "Consumer",
Description = "The consumer to connect to.",
UIHandler = typeof(ConsumerDefinitionsDropdownOptionsProvider),
UIHint = InputUIHints.DropDown
)]
public Input<string> ConsumerDefinitionId { get; set; } = default!;
/// <summary>
/// The topics to read from.
/// </summary>
[Input(
DisplayName = "Topics",
Description = "The topics to read from.",
UIHint = InputUIHints.MultiText
)]
public Input<ICollection<string>> Topics { get; set; } = default!;
/// <summary>
/// Optional. The .NET type to deserialize the message into. Defaults to <see cref="string"/>.
/// </summary>
[Input(Description = "Optional. The .NET type to deserialize the message into.")]
public Input<Type?> MessageType { get; set; } = default!;
[Input(
Description = "Optional. A predicate to filter messages.",
AutoEvaluate = false
)]
public Input<bool> Predicate { get; set; } = default!;
[Input(DisplayName = "Local", Description = "Whether the event is local to the workflow. When checked, only events delivered to this workflow instance will resume this activity.")]
public Input<bool> IsLocal { get; set; } = default!;
/// <summary>
/// The received transport message.
/// </summary>
public Output<KafkaTransportMessage> TransportMessage = default!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
// If the activity is triggered by a workflow trigger, resume immediately.
if (context.IsTriggerOfWorkflow())
{
await Resume(context);
}
else
{
// Otherwise, create a bookmark and wait for the stimulus to arrive.
context.CreateBookmark(GetStimulus(context.ExpressionExecutionContext), Resume, false);
}
}
/// <inheritdoc />
protected override object GetTriggerPayload(TriggerIndexingContext context) => GetStimulus(context.ExpressionExecutionContext);
private async ValueTask Resume(ActivityExecutionContext context)
{
var receivedMessage = context.GetWorkflowInput<KafkaTransportMessage>(InputKey);
SetResult(receivedMessage, context);
await context.CompleteActivityAsync();
}
private void SetResult(KafkaTransportMessage receivedMessage, ActivityExecutionContext context)
{
var bodyAsString = receivedMessage.Value;
var targetType = MessageType.GetOrDefault(context);
var deserializer = context.GetRequiredService<IOptions<KafkaOptions>>().Value.Deserializer;
var serviceProvider = context.WorkflowExecutionContext.ServiceProvider;
var body = targetType == null ? bodyAsString : deserializer(serviceProvider, bodyAsString, targetType);
context.Set(TransportMessage, receivedMessage);
context.SetResult(body);
}
private object GetStimulus(ExpressionExecutionContext context)
{
var consumerDefinitionId = ConsumerDefinitionId.Get(context);
var topics = Topics.GetOrDefault(context) ?? [];
var messageType = MessageType.GetOrDefault(context) ?? typeof(string);
var isLocal = IsLocal.GetOrDefault(context);
var activity = context.GetActivity();
var activityRegistry = context.GetRequiredService<IActivityRegistry>();
var activityDescriptor = activityRegistry.Find(activity.Type, activity.Version)!;
var inputDescriptor = activityDescriptor.GetWrappedInputPropertyDescriptor(activity, nameof(Predicate));
var predicateInput = (Input?)inputDescriptor!.ValueGetter(activity);
var predicateExpression = predicateInput?.Expression;
return new MessageReceivedStimulus
{
ConsumerDefinitionId = consumerDefinitionId,
Topics = topics.Distinct().ToList(),
MessageType = messageType,
IsLocal = isLocal,
Predicate = predicateExpression,
};
}
}

102
src/modules/Elsa.Kafka/Activities/SendMessage.cs Normal file
View file

@ -0,0 +1,102 @@
using System.Text;
using Confluent.Kafka;
using Elsa.Extensions;
using Elsa.Kafka.UIHints;
using Elsa.Workflows;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.UIHints;
using Microsoft.Extensions.Options;
namespace Elsa.Kafka.Activities;
[Activity("Elsa.Kafka", "Kafka", "Sends a message to a given topic")]
public class SendMessage : CodeActivity
{
/// <summary>
/// The topic to which the message will be sent.
/// </summary>
[Input(
Description = "The topic to which the message will be sent.",
UIHint = InputUIHints.DropDown,
UIHandler = typeof(TopicDefinitionsDropdownOptionsProvider)
)]
public Input<string> Topic { get; set; } = default!;
/// <summary>
/// The producer to use when sending the message.
/// </summary>
[Input(
DisplayName = "Producer",
Description = "The producer to use when sending the message.",
UIHint = InputUIHints.DropDown,
UIHandler = typeof(ProducerDefinitionsDropdownOptionsProvider)
)]
public Input<string> ProducerDefinitionId { get; set; } = default!;
[Input(DisplayName = "Local", Description = "When checked, the message will be delivered to this workflow instance only.")]
public Input<bool> IsLocal { get; set; } = default!;
/// <summary>
/// Optional. The correlation ID to assign to the message.
/// </summary>
[Input(
DisplayName = "Correlation ID",
Description = "Optional. The correlation ID to assign to the message."
)]
public Input<string?> CorrelationId { get; set; } = default!;
/// <summary>
/// The content of the message to send.
/// </summary>
[Input(Description = "The content of the message to send.")]
public Input<object> Content { get; set; } = default!;
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var topic = Topic.Get(context);
var producerDefinitionId = ProducerDefinitionId.Get(context);
var producerDefinitionEnumerator = context.GetRequiredService<IProducerDefinitionEnumerator>();
var producerDefinition = await producerDefinitionEnumerator.GetByIdAsync(producerDefinitionId);
var content = Content.Get(context);
var serializer = context.GetRequiredService<IOptions<KafkaOptions>>().Value.Serializer;
var serviceProvider = context.WorkflowExecutionContext.ServiceProvider;
var serializedContent = content as string ?? serializer(serviceProvider, content);
var config = new ProducerConfig
{
BootstrapServers = string.Join(",", producerDefinition.BootstrapServers),
};
context.DeferTask(async () =>
{
using var producer = new ProducerBuilder<Null, string>(config).Build();
var headers = CreateHeaders(context);
var message = new Message<Null, string>
{
Value = serializedContent,
Headers = headers
};
await producer.ProduceAsync(topic, message);
});
}
private Headers CreateHeaders(ActivityExecutionContext context)
{
var options = context.GetRequiredService<IOptions<KafkaOptions>>().Value;
var headers = new Headers();
var correlationId = CorrelationId.GetOrDefault(context);
var isLocal = IsLocal.Get(context);
var topic = Topic.Get(context);
headers.Add(options.TopicHeaderKey, Encoding.UTF8.GetBytes(topic));
if (!string.IsNullOrWhiteSpace(correlationId))
headers.Add(options.CorrelationHeaderKey, Encoding.UTF8.GetBytes(correlationId));
if (isLocal)
headers.Add(options.WorkflowInstanceIdHeaderKey, Encoding.UTF8.GetBytes(context.WorkflowExecutionContext.Id));
return headers;
}
}

11
src/modules/Elsa.Kafka/Contracts/IConsumerDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,11 @@
namespace Elsa.Kafka;
public interface IConsumerDefinitionEnumerator
{
/// <summary>
/// Retrieves a list of consumer definitions provided by <see cref="IConsumerDefinitionProvider"/> implementations.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A list of ConsumerDefinition objects.</returns>
Task<IEnumerable<ConsumerDefinition>> EnumerateAsync(CancellationToken cancellationToken);
}

6
src/modules/Elsa.Kafka/Contracts/IConsumerDefinitionProvider.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Kafka;
public interface IConsumerDefinitionProvider
{
Task<IEnumerable<ConsumerDefinition>> GetConsumerDefinitionsAsync(CancellationToken cancellationToken = default);
}

9
src/modules/Elsa.Kafka/Contracts/ICorrelationStrategy.cs Normal file
View file

@ -0,0 +1,9 @@
namespace Elsa.Kafka;
/// <summary>
/// Interface representing a strategy for extracting a correlation ID from a <see cref="KafkaTransportMessage"/>.
/// </summary>
public interface ICorrelationStrategy
{
string? GetCorrelationId(KafkaTransportMessage transportMessage);
}

16
src/modules/Elsa.Kafka/Contracts/IProducerDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,16 @@
namespace Elsa.Kafka;
public interface IProducerDefinitionEnumerator
{
/// <summary>
/// Retrieves a list of producer definitions provided by <see cref="IProducerDefinitionProvider"/> implementations.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A list of ProducerDefinition objects.</returns>
Task<IEnumerable<ProducerDefinition>> EnumerateAsync(CancellationToken cancellationToken);
/// <summary>
/// Retrieves a producer definition by its ID.
/// </summary>
Task<ProducerDefinition> GetByIdAsync(string id);
}

6
src/modules/Elsa.Kafka/Contracts/IProducerDefinitionProvider.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Kafka;
public interface IProducerDefinitionProvider
{
Task<IEnumerable<ProducerDefinition>> GetProducerDefinitionsAsync(CancellationToken cancellationToken = default);
}

11
src/modules/Elsa.Kafka/Contracts/ITopicDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,11 @@
namespace Elsa.Kafka;
public interface ITopicDefinitionEnumerator
{
/// <summary>
/// Retrieves a list of topic definitions provided by <see cref="ITopicDefinitionProvider"/> implementations.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A list of TopicDefinition objects.</returns>
Task<IEnumerable<TopicDefinition>> EnumerateAsync(CancellationToken cancellationToken);
}

6
src/modules/Elsa.Kafka/Contracts/ITopicDefinitionProvider.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Kafka;
public interface ITopicDefinitionProvider
{
Task<IEnumerable<TopicDefinition>> GetTopicDefinitionsAsync(CancellationToken cancellationToken = default);
}

11
src/modules/Elsa.Kafka/Contracts/IWorker.cs Normal file
View file

@ -0,0 +1,11 @@
namespace Elsa.Kafka;
public interface IWorker : IDisposable
{
void Start(CancellationToken cancellationToken);
void Stop();
void BindTrigger(TriggerBinding binding);
void BindBookmark(BookmarkBinding binding);
void RemoveTriggers(IEnumerable<string> triggerIds);
void RemoveBookmarks(IEnumerable<string> bookmarkIds);
}

12
src/modules/Elsa.Kafka/Contracts/IWorkerManager.cs Normal file
View file

@ -0,0 +1,12 @@
using Elsa.Workflows.Runtime.Entities;
namespace Elsa.Kafka;
public interface IWorkerManager
{
Task UpdateWorkersAsync(CancellationToken cancellationToken = default);
void StopWorkers();
Task BindTriggersAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default);
Task BindBookmarksAsync(IEnumerable<StoredBookmark> bookmarks, CancellationToken cancellationToken = default);
IWorker? GetWorker(string consumerDefinitionId);
}

9
src/modules/Elsa.Kafka/Contracts/IWorkerTopicSubscriber.cs Normal file
View file

@ -0,0 +1,9 @@
namespace Elsa.Kafka;
public interface IWorkerTopicSubscriber
{
/// <summary>
/// Updates all workers by subscribing to topics based on the current workflow trigger indexes and bookmarks.
/// </summary>
Task UpdateTopicSubscriptionsAsync(CancellationToken cancellationToken);
}

19
src/modules/Elsa.Kafka/Correlation/HeaderCorrelationStrategy.cs Normal file
View file

@ -0,0 +1,19 @@
using System.Text;
using Microsoft.Extensions.Options;
namespace Elsa.Kafka;
/// <summary>
/// A correlation strategy that retrieves a correlation ID from a specified header in a Kafka transport message.
/// </summary>
public class HeaderCorrelationStrategy(IOptions<KafkaOptions> options) : ICorrelationStrategy
{
/// <inheritdoc/>
public string? GetCorrelationId(KafkaTransportMessage transportMessage)
{
var headerName = options.Value.CorrelationHeaderKey;
return !transportMessage.Headers.TryGetValue(headerName, out var headerValue)
? null
: Encoding.UTF8.GetString(headerValue);
}
}

11
src/modules/Elsa.Kafka/Correlation/NullCorrelationStrategy.cs Normal file
View file

@ -0,0 +1,11 @@
namespace Elsa.Kafka;
/// <summary>
/// Represents a strategy that does not extract any correlation ID from a <see cref="KafkaTransportMessage"/>.
/// This implementation of <see cref="ICorrelationStrategy"/> always returns null,
/// effectively indicating no correlation ID is available or applicable.
/// </summary>
public class NullCorrelationStrategy : ICorrelationStrategy
{
public string? GetCorrelationId(KafkaTransportMessage transportMessage) => null;
}

24
src/modules/Elsa.Kafka/Elsa.Kafka.csproj Normal file
View file

@ -0,0 +1,24 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<Description>
Provides Apache Kafka integration and activities.
</Description>
<PackageTags>elsa module kafka service-bus</PackageTags>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Azure.Messaging.ServiceBus" />
<PackageReference Include="Azure.ResourceManager.ServiceBus" />
<PackageReference Include="Confluent.Kafka" />
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
<PackageReference Include="System.Linq.Async" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
</ItemGroup>
</Project>

9
src/modules/Elsa.Kafka/Elsa.Kafka.csproj.DotSettings Normal file
View file

@ -0,0 +1,9 @@
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=contracts/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=correlation/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=entities/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=extensions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=features/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=models/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=options/@EntryIndexedValue">True</s:Boolean>
</wpf:ResourceDictionary>

13
src/modules/Elsa.Kafka/Entities/ConsumerConfigDefinition.cs Normal file
View file

@ -0,0 +1,13 @@
using Confluent.Kafka;
using Elsa.Common.Entities;
namespace Elsa.Kafka;
public class ConsumerDefinition : Entity
{
public string Name { get; set; } = default!;
public ICollection<string> BootstrapServers { get; set; } = [];
public string GroupId { get; set; } = default!;
public AutoOffsetReset AutoOffsetReset { get; set; } = AutoOffsetReset.Earliest;
public bool EnableAutoCommit { get; set; }
}

9
src/modules/Elsa.Kafka/Entities/ProducerDefinition.cs Normal file
View file

@ -0,0 +1,9 @@
using Elsa.Common.Entities;
namespace Elsa.Kafka;
public class ProducerDefinition : Entity
{
public string Name { get; set; } = default!;
public ICollection<string> BootstrapServers { get; set; } = [];
}

11
src/modules/Elsa.Kafka/Entities/TopicDefinition.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Common.Entities;
namespace Elsa.Kafka;
public class TopicDefinition : Entity
{
/// <summary>
/// Gets or sets the name of the topic.
/// </summary>
public string Name { get; set; } = default!;
}

12
src/modules/Elsa.Kafka/Extensions/ModuleExtensions.cs Normal file
View file

@ -0,0 +1,12 @@
using Elsa.Extensions;
using Elsa.Features.Services;
namespace Elsa.Kafka;
public static class ModuleExtensions
{
public static IModule UseKafka(this IModule module, Action<KafkaFeature>? configure = null)
{
return module.Use(configure);
}
}

76
src/modules/Elsa.Kafka/Features/KafkaFeature.cs Normal file
View file

@ -0,0 +1,76 @@
using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Elsa.Kafka.Implementations;
using Elsa.Kafka.Providers;
using Elsa.Kafka.Tasks;
using Elsa.Kafka.UIHints;
using Elsa.Workflows;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Kafka;
public class KafkaFeature(IModule module) : FeatureBase(module)
{
private Action<KafkaOptions> _configureOptions = _ => { };
private Func<IServiceProvider, ICorrelationStrategy> _correlationStrategyFactory = sp => sp.GetRequiredService<HeaderCorrelationStrategy>();
public KafkaFeature ConfigureOptions(Action<KafkaOptions> configureOptions)
{
_configureOptions += configureOptions;
return this;
}
public KafkaFeature WithCorrelationStrategy<T>() where T : class, ICorrelationStrategy
{
Services.AddScoped<T>();
_correlationStrategyFactory = sp => sp.GetRequiredService<T>();
return this;
}
public KafkaFeature WithCorrelationStrategy(Func<IServiceProvider, ICorrelationStrategy> correlationStrategyFactory)
{
_correlationStrategyFactory = correlationStrategyFactory;
return this;
}
public KafkaFeature WithCorrelationStrategy(Func<ICorrelationStrategy> correlationStrategyFactory)
{
_correlationStrategyFactory = _ => correlationStrategyFactory();
return this;
}
public KafkaFeature WithCorrelationStrategy(ICorrelationStrategy correlationStrategy)
{
_correlationStrategyFactory = _ => correlationStrategy;
return this;
}
public override void Configure()
{
Module.AddActivitiesFrom<KafkaFeature>();
}
public override void Apply()
{
Services.Configure(_configureOptions);
Services
.AddBackgroundTask<StartConsumersStartupTask>()
.AddSingleton<IWorkerManager, WorkerManager>()
.AddScoped<IWorkerTopicSubscriber, WorkerTopicSubscriber>()
.AddScoped<IConsumerDefinitionProvider, OptionsDefinitionProvider>()
.AddScoped<IProducerDefinitionProvider, OptionsDefinitionProvider>()
.AddScoped<ITopicDefinitionProvider, OptionsDefinitionProvider>()
.AddScoped<IConsumerDefinitionEnumerator, ConsumerDefinitionEnumerator>()
.AddScoped<IProducerDefinitionEnumerator, ProducerDefinitionEnumerator>()
.AddScoped<ITopicDefinitionEnumerator, TopicDefinitionEnumerator>()
.AddScoped<IPropertyUIHandler, ConsumerDefinitionsDropdownOptionsProvider>()
.AddScoped<IPropertyUIHandler, ProducerDefinitionsDropdownOptionsProvider>()
.AddScoped<IPropertyUIHandler, TopicDefinitionsDropdownOptionsProvider>()
.AddScoped<HeaderCorrelationStrategy>()
.AddScoped<NullCorrelationStrategy>()
.AddScoped(_correlationStrategyFactory)
.AddHandlersFrom<KafkaFeature>();
}
}

3
src/modules/Elsa.Kafka/FodyWeavers.xml Normal file
View file

@ -0,0 +1,3 @@
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
<ConfigureAwait />
</Weavers>

208
src/modules/Elsa.Kafka/Handlers/TriggerWorkflows.cs Normal file
View file

@ -0,0 +1,208 @@
using System.Text;
using Elsa.Expressions.Contracts;
using Elsa.Expressions.Models;
using Elsa.Kafka.Activities;
using Elsa.Kafka.Notifications;
using Elsa.Kafka.Stimuli;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Memory;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Options;
using JetBrains.Annotations;
using Microsoft.Extensions.Options;
namespace Elsa.Kafka.Handlers;
[UsedImplicitly]
public class TriggerWorkflows(
ITriggerInvoker triggerInvoker,
IBookmarkQueue bookmarkQueue,
ICorrelationStrategy correlationStrategy,
IExpressionEvaluator expressionEvaluator,
IOptions<KafkaOptions> options,
IServiceProvider serviceProvider) : INotificationHandler<TransportMessageReceived>
{
private static readonly string MessageReceivedActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<MessageReceived>();
public async Task HandleAsync(TransportMessageReceived notification, CancellationToken cancellationToken)
{
var worker = notification.Worker;
var consumerDefinition = worker.ConsumerDefinition;
var consumerDefinitionId = consumerDefinition.Id;
var transportMessage = notification.TransportMessage;
var boundTriggers = worker.TriggerBindings.Values;
var boundBookmarks = worker.BookmarkBindings.Values;
var matchingTriggers = await GetMatchingTriggerBindingsAsync(boundTriggers, consumerDefinitionId, transportMessage, cancellationToken);
var matchingBookmarks = await GetMatchingBookmarkBindingsAsync(boundBookmarks, consumerDefinitionId, transportMessage, cancellationToken);
await InvokeTriggersAsync(matchingTriggers, transportMessage, cancellationToken);
await InvokeBookmarksAsync(matchingBookmarks, transportMessage, cancellationToken);
}
private async Task InvokeTriggersAsync(IEnumerable<TriggerBinding> matchingTriggers, KafkaTransportMessage transportMessage, CancellationToken cancellationToken)
{
var input = new Dictionary<string, object>
{
[MessageReceived.InputKey] = transportMessage
};
foreach (var binding in matchingTriggers)
{
var invokeTriggerRequest = new InvokeTriggerRequest
{
Workflow = binding.Workflow,
ActivityId = binding.TriggerActivityId,
CorrelationId = GetCorrelationId(transportMessage),
Input = input,
Properties = new Dictionary<string, object>
{
[MessageReceived.InputKey] = transportMessage
}
};
await triggerInvoker.InvokeAsync(invokeTriggerRequest, cancellationToken);
}
}
private async Task InvokeBookmarksAsync(IEnumerable<BookmarkBinding> matchingBookmarks, KafkaTransportMessage transportMessage, CancellationToken cancellationToken)
{
var input = new Dictionary<string, object>
{
[MessageReceived.InputKey] = transportMessage
};
var properties = new Dictionary<string, object>
{
[MessageReceived.InputKey] = transportMessage
};
foreach (var binding in matchingBookmarks)
{
var bookmarkQueueItem = new NewBookmarkQueueItem
{
WorkflowInstanceId = binding.WorkflowInstanceId,
BookmarkId = binding.BookmarkId,
Options = new ResumeBookmarkOptions
{
Input = input,
Properties = properties
},
ActivityTypeName = MessageReceivedActivityTypeName
};
await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken);
}
}
private async Task<IEnumerable<TriggerBinding>> GetMatchingTriggerBindingsAsync(
IEnumerable<TriggerBinding> boundTriggers,
string consumerDefinitionId,
KafkaTransportMessage transportMessage,
CancellationToken cancellationToken)
{
var matchingTriggers = new List<TriggerBinding>();
var topic = GetTopic(transportMessage);
if (string.IsNullOrEmpty(topic))
return matchingTriggers;
foreach (var binding in boundTriggers)
{
var stimulus = binding.Stimulus;
if (stimulus.ConsumerDefinitionId != consumerDefinitionId)
continue;
if (stimulus.Topics.All(x => x != topic))
continue;
var isMatch = await EvaluatePredicateAsync(transportMessage, stimulus, cancellationToken);
if (isMatch)
matchingTriggers.Add(binding);
}
return matchingTriggers;
}
private async Task<IEnumerable<BookmarkBinding>> GetMatchingBookmarkBindingsAsync(
IEnumerable<BookmarkBinding> boundBookmarks,
string consumerDefinitionId,
KafkaTransportMessage transportMessage,
CancellationToken cancellationToken)
{
var matchingBookmarks = new List<BookmarkBinding>();
var topic = GetTopic(transportMessage);
if (string.IsNullOrEmpty(topic))
return matchingBookmarks;
var correlationId = GetCorrelationId(transportMessage);
var workflowInstanceId = GetWorkflowInstanceId(transportMessage);
foreach (var binding in boundBookmarks)
{
var stimulus = binding.Stimulus;
if (stimulus.ConsumerDefinitionId != consumerDefinitionId)
continue;
if (stimulus.Topics.All(x => x != topic))
continue;
if (stimulus.IsLocal)
{
if (binding.WorkflowInstanceId != workflowInstanceId)
{
if (string.IsNullOrWhiteSpace(correlationId))
continue;
if (binding.CorrelationId != correlationId)
continue;
}
}
var isMatch = await EvaluatePredicateAsync(transportMessage, stimulus, cancellationToken);
if (isMatch)
matchingBookmarks.Add(binding);
}
return matchingBookmarks;
}
private async Task<bool> EvaluatePredicateAsync(KafkaTransportMessage transportMessage, MessageReceivedStimulus stimulus, CancellationToken cancellationToken)
{
var predicate = stimulus.Predicate;
if (predicate == null)
return true;
var memory = new MemoryRegister();
var messageVariable = new Variable("message", transportMessage);
var messageType = stimulus.MessageType;
var message = DeserializeMessage(transportMessage.Value, messageType);
var expressionExecutionContext = new ExpressionExecutionContext(serviceProvider, memory, cancellationToken: cancellationToken);
messageVariable.Set(expressionExecutionContext, message);
return await expressionEvaluator.EvaluateAsync<bool>(predicate, expressionExecutionContext);
}
private object DeserializeMessage(string value, Type? type)
{
return type == null ? value : options.Value.Deserializer(serviceProvider, value, type);
}
private string? GetWorkflowInstanceId(KafkaTransportMessage transportMessage)
{
var key = options.Value.WorkflowInstanceIdHeaderKey;
return transportMessage.Headers.TryGetValue(key, out var value) ? Encoding.UTF8.GetString(value) : null;
}
private string? GetCorrelationId(KafkaTransportMessage transportMessage)
{
return correlationStrategy.GetCorrelationId(transportMessage);
}
private string? GetTopic(KafkaTransportMessage transportMessage)
{
var key = options.Value.TopicHeaderKey;
return transportMessage.Headers.TryGetValue(key, out var value) ? Encoding.UTF8.GetString(value) : null;
}
}

102
src/modules/Elsa.Kafka/Handlers/UpdateWorkers.cs Normal file
View file

@ -0,0 +1,102 @@
using Elsa.Extensions;
using Elsa.Kafka.Activities;
using Elsa.Kafka.Stimuli;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Management;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime.Notifications;
using JetBrains.Annotations;
namespace Elsa.Kafka.Handlers;
/// <summary>
/// Creates workers for each trigger &amp; bookmark in response to updated workflow trigger indexes and bookmarks.
/// </summary>
[UsedImplicitly]
public class UpdateWorkers(IWorkerManager workerManager, IWorkflowDefinitionService workflowDefinitionService) :
INotificationHandler<WorkflowTriggersIndexed>,
INotificationHandler<WorkflowBookmarksIndexed>,
INotificationHandler<BookmarksDeleted>
{
private static readonly string MessageReceivedActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<MessageReceived>();
/// Adds, updates and removes workers based on added and removed triggers.
public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken)
{
var removedTriggers = notification.IndexedWorkflowTriggers.RemovedTriggers.Where(x => x.Name == MessageReceivedActivityTypeName).ToList();
var removedTriggerIds = removedTriggers.Select(x => x.Id).ToList();
foreach (var removedTrigger in removedTriggers)
{
var consumerDefinitionId = removedTrigger.GetPayload<MessageReceivedStimulus>().ConsumerDefinitionId;
var worker = workerManager.GetWorker(consumerDefinitionId);
worker?.RemoveTriggers(removedTriggerIds);
}
var addedTriggers = notification.IndexedWorkflowTriggers.AddedTriggers.Where(x => x.Name == MessageReceivedActivityTypeName).ToList();
foreach (var addedTrigger in addedTriggers)
{
var stimulus = addedTrigger.GetPayload<MessageReceivedStimulus>();
var consumerDefinitionId = stimulus.ConsumerDefinitionId;
var worker = workerManager.GetWorker(consumerDefinitionId);
if (worker == null)
continue;
var workflow = await workflowDefinitionService.FindWorkflowAsync(addedTrigger.WorkflowDefinitionVersionId, cancellationToken);
if (workflow == null)
continue;
var triggerBinding = new TriggerBinding(workflow, addedTrigger.Id, addedTrigger.ActivityId, stimulus);
worker.BindTrigger(triggerBinding);
}
}
/// Adds, updates and removes workers based on added and removed bookmarks.
public Task HandleAsync(WorkflowBookmarksIndexed notification, CancellationToken cancellationToken)
{
RemoveBookmarkBindings(notification.IndexedWorkflowBookmarks.RemovedBookmarks.Where(x => x.Name == MessageReceivedActivityTypeName).ToList());
var addedBookmarks = notification.IndexedWorkflowBookmarks.AddedBookmarks.Where(x => x.Name == MessageReceivedActivityTypeName).ToList();
var workflowInstanceId = notification.IndexedWorkflowBookmarks.WorkflowExecutionContext.Id;
var correlationId = notification.IndexedWorkflowBookmarks.WorkflowExecutionContext.CorrelationId;
foreach (var addedBookmark in addedBookmarks)
{
var stimulus = addedBookmark.GetPayload<MessageReceivedStimulus>();
var consumerDefinitionId = stimulus.ConsumerDefinitionId;
var worker = workerManager.GetWorker(consumerDefinitionId);
if (worker == null)
continue;
var bookmarkBinding = new BookmarkBinding(workflowInstanceId, correlationId, addedBookmark.Id, stimulus);
worker.BindBookmark(bookmarkBinding);
}
return Task.CompletedTask;
}
public Task HandleAsync(BookmarksDeleted notification, CancellationToken cancellationToken)
{
RemoveBookmarkBindings(notification.Bookmarks.Where(x => x.ActivityTypeName == MessageReceivedActivityTypeName).Select(x => x.ToBookmark()).ToList());
return Task.CompletedTask;
}
private void RemoveBookmarkBindings(ICollection<Bookmark> bookmarks)
{
var removedBookmarkIds = bookmarks.Select(x => x.Id).ToList();
foreach (var bookmark in bookmarks)
{
var stimulus = bookmark.GetPayload<MessageReceivedStimulus>();
var consumerDefinitionId = stimulus.ConsumerDefinitionId;
var worker = workerManager.GetWorker(consumerDefinitionId);
worker?.RemoveBookmarks(removedBookmarkIds);
}
}
}

23
src/modules/Elsa.Kafka/Implementations/ConsumerDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,23 @@
using System.Runtime.CompilerServices;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.Implementations;
public class ConsumerDefinitionEnumerator(IEnumerable<IConsumerDefinitionProvider> providers) : IConsumerDefinitionEnumerator
{
public async Task<IEnumerable<ConsumerDefinition>> EnumerateAsync(CancellationToken cancellationToken)
{
return await GetConsumerDefinitionsInternalAsync(cancellationToken).ToListAsync(cancellationToken);
}
private async IAsyncEnumerable<ConsumerDefinition> GetConsumerDefinitionsInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (var provider in providers)
{
var consumerDefinitions = await provider.GetConsumerDefinitionsAsync(cancellationToken).ToList();
foreach (var consumerDefinition in consumerDefinitions)
yield return consumerDefinition;
}
}
}

29
src/modules/Elsa.Kafka/Implementations/ProducerDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,29 @@
using System.Runtime.CompilerServices;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.Implementations;
public class ProducerDefinitionEnumerator(IEnumerable<IProducerDefinitionProvider> providers) : IProducerDefinitionEnumerator
{
public async Task<IEnumerable<ProducerDefinition>> EnumerateAsync(CancellationToken cancellationToken)
{
return await GetProducerDefinitionsInternalAsync(cancellationToken).ToListAsync(cancellationToken);
}
public async Task<ProducerDefinition> GetByIdAsync(string id)
{
var producerDefinitions = await GetProducerDefinitionsInternalAsync(CancellationToken.None).ToListAsync();
return producerDefinitions.FirstOrDefault(x => x.Id == id) ?? throw new InvalidOperationException($"Producer definition with ID '{id}' not found.");
}
private async IAsyncEnumerable<ProducerDefinition> GetProducerDefinitionsInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (var provider in providers)
{
var producerDefinitions = await provider.GetProducerDefinitionsAsync(cancellationToken).ToList();
foreach (var producerDefinition in producerDefinitions)
yield return producerDefinition;
}
}
}

23
src/modules/Elsa.Kafka/Implementations/TopicDefinitionEnumerator.cs Normal file
View file

@ -0,0 +1,23 @@
using System.Runtime.CompilerServices;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.Implementations;
public class TopicDefinitionEnumerator(IEnumerable<ITopicDefinitionProvider> providers) : ITopicDefinitionEnumerator
{
public async Task<IEnumerable<TopicDefinition>> EnumerateAsync(CancellationToken cancellationToken)
{
return await GetTopicDefinitionsInternalAsync(cancellationToken).ToListAsync(cancellationToken);
}
private async IAsyncEnumerable<TopicDefinition> GetTopicDefinitionsInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (var provider in providers)
{
var topicDefinitions = await provider.GetTopicDefinitionsAsync(cancellationToken).ToList();
foreach (var topicDefinition in topicDefinitions)
yield return topicDefinition;
}
}
}

128
src/modules/Elsa.Kafka/Implementations/Worker.cs Normal file
View file

@ -0,0 +1,128 @@
using Confluent.Kafka;
using Elsa.Extensions;
namespace Elsa.Kafka.Implementations;
public class Worker(ConsumerDefinition consumerDefinition) : IWorker
{
private bool _running;
private IConsumer<Ignore, string> _consumer = default!;
private CancellationTokenSource _cancellationTokenSource = new();
private readonly HashSet<string> _subscribedTopics = new();
public IDictionary<string, BookmarkBinding> BookmarkBindings { get; } = new Dictionary<string, BookmarkBinding>();
public IDictionary<string, TriggerBinding> TriggerBindings { get; } = new Dictionary<string, TriggerBinding>();
public Func<Worker, Message<Ignore, string>, CancellationToken, Task>? MessageReceived { get; set; }
public ConsumerDefinition ConsumerDefinition { get; } = consumerDefinition;
public void Start(CancellationToken cancellationToken)
{
if (_running)
return;
_running = true;
var config = new ConsumerConfig
{
BootstrapServers = string.Join(",", ConsumerDefinition.BootstrapServers),
GroupId = ConsumerDefinition.GroupId,
AutoOffsetReset = ConsumerDefinition.AutoOffsetReset,
EnableAutoCommit = ConsumerDefinition.EnableAutoCommit
};
_consumer = new ConsumerBuilder<Ignore, string>(config).Build();
_cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
_ = Task.Run(() => RunAsync(_cancellationTokenSource.Token), cancellationToken);
}
public void Stop()
{
if (!_running)
return;
_running = false;
_consumer.Unsubscribe();
_consumer.Close();
_consumer.Dispose();
_cancellationTokenSource.Cancel();
_cancellationTokenSource.Dispose();
}
public void BindTrigger(TriggerBinding binding)
{
var topics = binding.Stimulus.Topics.Distinct().ToList();
TriggerBindings[binding.TriggerId] = binding;
Subscribe(topics);
}
public void BindBookmark(BookmarkBinding binding)
{
var topics = binding.Stimulus.Topics.Distinct().ToList();
BookmarkBindings[binding.BookmarkId] = binding;
Subscribe(topics);
}
public void RemoveTriggers(IEnumerable<string> triggerIds)
{
var triggerIdList = triggerIds.ToList();
TriggerBindings.RemoveWhere(x => triggerIdList.Contains(x.Key));
}
public void RemoveBookmarks(IEnumerable<string> bookmarkIds)
{
var bookmarkIdList = bookmarkIds.ToList();
BookmarkBindings.RemoveWhere(x => bookmarkIdList.Contains(x.Key));
}
public void Subscribe(IEnumerable<BookmarkBinding> bookmarkBindings)
{
var topics = bookmarkBindings.SelectMany(x => x.Stimulus.Topics).Distinct().ToList();
Subscribe(topics);
}
public void Dispose()
{
Stop();
}
private void Subscribe(IEnumerable<string> topics)
{
if(!_running)
throw new InvalidOperationException("The worker is not running.");
var topicList = topics.ToHashSet();
// Check if there are any topics not yet in the list of subscribed topics.
if (topicList.All(topic => _subscribedTopics.Contains(topic)))
return;
// Add the new topics to the list of subscribed topics.
_subscribedTopics.UnionWith(topicList);
// Update the consumer's subscription.
_consumer.Subscribe(_subscribedTopics);
}
private async Task RunAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
var consumeResult = _consumer.Consume(cancellationToken);
if (consumeResult.IsPartitionEOF)
continue;
await ProcessMessageAsync(consumeResult.Message);
}
_consumer.Unsubscribe();
_consumer.Close();
_consumer.Dispose();
}
private Task ProcessMessageAsync(Message<Ignore, string> message)
{
MessageReceived?.Invoke(this, message, _cancellationTokenSource.Token);
return Task.CompletedTask;
}
}

136
src/modules/Elsa.Kafka/Implementations/WorkerManager.cs Normal file
View file

@ -0,0 +1,136 @@
using Confluent.Kafka;
using Elsa.Extensions;
using Elsa.Kafka.Notifications;
using Elsa.Kafka.Stimuli;
using Elsa.Mediator.Contracts;
using Elsa.Workflows;
using Elsa.Workflows.Management;
using Elsa.Workflows.Runtime.Entities;
using Microsoft.Extensions.DependencyInjection;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.Implementations;
public class WorkerManager(IHasher hasher, IServiceScopeFactory scopeFactory) : IWorkerManager
{
private IDictionary<string, Worker> Workers { get; set; } = new Dictionary<string, Worker>();
public async Task UpdateWorkersAsync(CancellationToken cancellationToken = default)
{
await using var scope = scopeFactory.CreateAsyncScope();
var consumerDefinitionEnumerator = scope.ServiceProvider.GetRequiredService<IConsumerDefinitionEnumerator>();
var consumerDefinitions = await consumerDefinitionEnumerator.EnumerateAsync(cancellationToken).ToList();
var workers = new Dictionary<string, Worker>(Workers);
var workersToRemove = workers.Keys.Except(consumerDefinitions.Select(x => x.Id)).ToList();
// Remove workers that are no longer needed.
foreach (var workerToRemove in workersToRemove)
{
if (workers.TryGetValue(workerToRemove, out var worker))
{
worker.Stop();
workers.Remove(workerToRemove);
}
}
// Add or update workers.
foreach (var consumerDefinition in consumerDefinitions)
{
var existingWorker = workers.GetValueOrDefault(consumerDefinition.Id);
if (existingWorker == null)
{
// Add a new worker.
var worker = CreateWorker(consumerDefinition);
workers.Add(consumerDefinition.Id, worker);
}
else
{
// Compare the existing worker's consumer definition with the new consumer definition using a hash.
var existingHash = ComputeHash(existingWorker.ConsumerDefinition);
var hash = ComputeHash(consumerDefinition);
// If the hash is different, update the worker.
if (existingHash != hash)
{
existingWorker.Stop();
var worker = CreateWorker(consumerDefinition);
workers[consumerDefinition.Id] = worker;
}
}
}
// Ensure all workers are running.
foreach (var worker in workers.Values)
worker.Start(cancellationToken);
// Update the worker dictionary.
Workers = workers;
}
public async Task BindTriggersAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default)
{
await using var scope = scopeFactory.CreateAsyncScope();
var workflowDefinitionService = scope.ServiceProvider.GetRequiredService<IWorkflowDefinitionService>();
foreach (var trigger in triggers)
{
var workflow = await workflowDefinitionService.FindWorkflowAsync(trigger.WorkflowDefinitionVersionId, cancellationToken);
if (workflow == null)
continue;
var stimulus = trigger.GetPayload<MessageReceivedStimulus>();
var consumerDefinitionId = stimulus.ConsumerDefinitionId;
var worker = GetWorker(consumerDefinitionId);
var triggerBinding = new TriggerBinding(workflow, trigger.Id, trigger.ActivityId, stimulus);
worker.BindTrigger(triggerBinding);
}
}
public Task BindBookmarksAsync(IEnumerable<StoredBookmark> bookmarks, CancellationToken cancellationToken = default)
{
foreach (var bookmark in bookmarks)
{
var stimulus = bookmark.GetPayload<MessageReceivedStimulus>();
var consumerDefinitionId = stimulus.ConsumerDefinitionId;
var worker = GetWorker(consumerDefinitionId);
var bookmarkBinding = new BookmarkBinding(bookmark.WorkflowInstanceId, bookmark.CorrelationId, bookmark.Id, stimulus);
worker.BindBookmark(bookmarkBinding);
}
return Task.CompletedTask;
}
public void StopWorkers()
{
foreach (var worker in Workers.Values)
worker.Stop();
}
public IWorker? GetWorker(string consumerDefinitionId)
{
return Workers.TryGetValue(consumerDefinitionId, out var worker) ? worker : null;
}
private Worker CreateWorker(ConsumerDefinition consumerDefinition)
{
var worker = new Worker(consumerDefinition);
worker.MessageReceived = OnMessageReceivedAsync;
return worker;
}
private async Task OnMessageReceivedAsync(Worker worker, Message<Ignore, string> arg, CancellationToken cancellationToken)
{
var headers = arg.Headers.ToDictionary(x => x.Key, x => x.GetValueBytes());
var notification = new TransportMessageReceived(worker, new KafkaTransportMessage(arg.Key, arg.Value, headers));
await using var scope = scopeFactory.CreateAsyncScope();
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
await mediator.SendAsync(notification, cancellationToken);
}
private string ComputeHash(ConsumerDefinition consumerDefinition)
{
return hasher.Hash(consumerDefinition);
}
}

38
src/modules/Elsa.Kafka/Implementations/WorkerTopicSubscriber.cs Normal file
View file

@ -0,0 +1,38 @@
using Elsa.Kafka.Activities;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
namespace Elsa.Kafka.Implementations;
public class WorkerTopicSubscriber(ITriggerStore triggerStore, IBookmarkStore bookmarkStore, IWorkerManager workerManager) : IWorkerTopicSubscriber
{
private static readonly string MessageReceivedActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<MessageReceived>();
public async Task UpdateTopicSubscriptionsAsync(CancellationToken cancellationToken)
{
var triggers = await GetStoredTriggersAsync(cancellationToken);
var bookmarks = await GetStoredBookmarksAsync(cancellationToken);
await workerManager.BindTriggersAsync(triggers, cancellationToken);
await workerManager.BindBookmarksAsync(bookmarks, cancellationToken);
}
private async Task<IEnumerable<StoredTrigger>> GetStoredTriggersAsync(CancellationToken cancellationToken)
{
var triggerFilter = new TriggerFilter
{
Name = MessageReceivedActivityTypeName
};
return await triggerStore.FindManyAsync(triggerFilter, cancellationToken);
}
private async Task<IEnumerable<StoredBookmark>> GetStoredBookmarksAsync(CancellationToken cancellationToken)
{
var bookmarkFilter = new BookmarkFilter
{
ActivityTypeName = MessageReceivedActivityTypeName
};
return await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken);
}
}

5
src/modules/Elsa.Kafka/Models/BookmarkBinding.cs Normal file
View file

@ -0,0 +1,5 @@
using Elsa.Kafka.Stimuli;
namespace Elsa.Kafka;
public record BookmarkBinding(string WorkflowInstanceId, string? CorrelationId, string BookmarkId, MessageReceivedStimulus Stimulus);

3
src/modules/Elsa.Kafka/Models/KafkaTransportMessage.cs Normal file
View file

@ -0,0 +1,3 @@
namespace Elsa.Kafka;
public record KafkaTransportMessage(object? Key, string Value, IDictionary<string, byte[]> Headers);

6
src/modules/Elsa.Kafka/Models/TriggerBinding.cs Normal file
View file

@ -0,0 +1,6 @@
using Elsa.Kafka.Stimuli;
using Elsa.Workflows.Activities;
namespace Elsa.Kafka;
public record TriggerBinding(Workflow Workflow, string TriggerId, string TriggerActivityId, MessageReceivedStimulus Stimulus);

6
src/modules/Elsa.Kafka/Notifications/TransportMessageReceived.cs Normal file
View file

@ -0,0 +1,6 @@
using Elsa.Kafka.Implementations;
using Elsa.Mediator.Contracts;
namespace Elsa.Kafka.Notifications;
public record TransportMessageReceived(Worker Worker, KafkaTransportMessage TransportMessage) : INotification;

16
src/modules/Elsa.Kafka/Options/KafkaOptions.cs Normal file
View file

@ -0,0 +1,16 @@
using Elsa.Kafka.Serialization;
namespace Elsa.Kafka;
public class KafkaOptions
{
public ICollection<TopicDefinition> Topics { get; set; } = [];
public ICollection<ConsumerDefinition> Consumers { get; set; } = [];
public ICollection<ProducerDefinition> Producers { get; set; } = [];
public string WorkflowInstanceIdHeaderKey { get; set; } = "x-workflow-instance-id";
public string CorrelationHeaderKey { get; set; } = "x-correlation-id";
public string TopicHeaderKey { get; set; } = "x-topic";
public Func<IServiceProvider, object, string> Serializer { get; set; } = DefaultSerializers.SerializePayload;
public Func<IServiceProvider, string, Type, object> Deserializer { get; set; } = DefaultSerializers.DeserializePayload;
}

21
src/modules/Elsa.Kafka/Providers/OptionsDefinitionProvider.cs Normal file
View file

@ -0,0 +1,21 @@
using Microsoft.Extensions.Options;
namespace Elsa.Kafka.Providers;
public class OptionsDefinitionProvider(IOptions<KafkaOptions> options) : IConsumerDefinitionProvider, IProducerDefinitionProvider, ITopicDefinitionProvider
{
public Task<IEnumerable<ConsumerDefinition>> GetConsumerDefinitionsAsync(CancellationToken cancellationToken = default)
{
return Task.FromResult<IEnumerable<ConsumerDefinition>>(options.Value.Consumers);
}
public Task<IEnumerable<ProducerDefinition>> GetProducerDefinitionsAsync(CancellationToken cancellationToken = default)
{
return Task.FromResult<IEnumerable<ProducerDefinition>>(options.Value.Producers);
}
public Task<IEnumerable<TopicDefinition>> GetTopicDefinitionsAsync(CancellationToken cancellationToken = default)
{
return Task.FromResult<IEnumerable<TopicDefinition>>(options.Value.Topics);
}
}

35
src/modules/Elsa.Kafka/Serialization/DefaultSerializers.cs Normal file
View file

@ -0,0 +1,35 @@
using System.Text.Json;
using Elsa.Workflows.Serialization.Converters;
namespace Elsa.Kafka.Serialization;
public static class DefaultSerializers
{
private static readonly JsonSerializerOptions SerializerOptions = GetOptions();
private static JsonSerializerOptions GetOptions()
{
var options = new JsonSerializerOptions
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
Converters =
{
new ExpandoObjectConverterFactory()
}
};
return options;
}
public static string SerializePayload(IServiceProvider serviceProvider, object obj)
{
var json = JsonSerializer.Serialize(obj, obj.GetType(), SerializerOptions);
return json;
}
public static object DeserializePayload(IServiceProvider serviceProvider, string json, Type type)
{
var value = JsonSerializer.Deserialize(json, type, SerializerOptions)!;
return value;
}
}

14
src/modules/Elsa.Kafka/Stimuli/MessageReceivedStimulus.cs Normal file
View file

@ -0,0 +1,14 @@
using Elsa.Expressions.Models;
using Elsa.Workflows.Attributes;
namespace Elsa.Kafka.Stimuli;
public class MessageReceivedStimulus
{
public string ConsumerDefinitionId { get; set; } = default!;
[ExcludeFromHash] public Type? MessageType { get; set; }
public ICollection<string> Topics { get; set; } = [];
public IDictionary<string, object?>? CorrelatingFields { get; set; }
public Expression? Predicate { get; set; }
public bool IsLocal { get; set; }
}

20
src/modules/Elsa.Kafka/Tasks/StartConsumersTask.cs Normal file
View file

@ -0,0 +1,20 @@
using Elsa.Common;
using JetBrains.Annotations;
namespace Elsa.Kafka.Tasks;
[UsedImplicitly]
public class StartConsumersStartupTask(IWorkerManager workerManager, IWorkerTopicSubscriber workerTopicSubscriber) : BackgroundTask
{
public override async Task StartAsync(CancellationToken cancellationToken)
{
await workerManager.UpdateWorkersAsync(cancellationToken);
await workerTopicSubscriber.UpdateTopicSubscriptionsAsync(cancellationToken);
}
public override Task StopAsync(CancellationToken cancellationToken)
{
workerManager.StopWorkers();
return Task.CompletedTask;
}
}

14
src/modules/Elsa.Kafka/UIHints/ConsumerDefinitionsDropdownOptionsProvider.cs Normal file
View file

@ -0,0 +1,14 @@
using System.Reflection;
using Elsa.Workflows.UIHints.Dropdown;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.UIHints;
public class ConsumerDefinitionsDropdownOptionsProvider(IConsumerDefinitionEnumerator consumerEnumerator) : DropDownOptionsProviderBase
{
protected override async ValueTask<ICollection<SelectListItem>> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken)
{
var definitions = await consumerEnumerator.EnumerateAsync(cancellationToken).ToList();
return definitions.Select(x => new SelectListItem(x.Name, x.Id)).ToList();
}
}

14
src/modules/Elsa.Kafka/UIHints/ProducerDefinitionsDropdownOptionsProvider.cs Normal file
View file

@ -0,0 +1,14 @@
using System.Reflection;
using Elsa.Workflows.UIHints.Dropdown;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.UIHints;
public class ProducerDefinitionsDropdownOptionsProvider(IProducerDefinitionEnumerator producerEnumerator) : DropDownOptionsProviderBase
{
protected override async ValueTask<ICollection<SelectListItem>> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken)
{
var definitions = await producerEnumerator.EnumerateAsync(cancellationToken).ToList();
return definitions.Select(x => new SelectListItem(x.Name, x.Id)).ToList();
}
}

14
src/modules/Elsa.Kafka/UIHints/TopicDefinitionsDropdownOptionsProvider.cs Normal file
View file

@ -0,0 +1,14 @@
using System.Reflection;
using Elsa.Workflows.UIHints.Dropdown;
using Open.Linq.AsyncExtensions;
namespace Elsa.Kafka.UIHints;
public class TopicDefinitionsDropdownOptionsProvider(ITopicDefinitionEnumerator topicEnumerator) : DropDownOptionsProviderBase
{
protected override async ValueTask<ICollection<SelectListItem>> GetItemsAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken)
{
var definitions = await topicEnumerator.EnumerateAsync(cancellationToken).ToList();
return definitions.Select(x => new SelectListItem(x.Name, x.Id)).ToList();
}
}

1
src/modules/Elsa.MassTransit/Services/MassTransitActivityTypeProvider.cs
View file

@ -61,6 +61,7 @@ public class MassTransitActivityTypeProvider(IActivityFactory activityFactory, I
return new()
{
Name = typeName,
TypeName = fullTypeName,
Version = 1,
DisplayName = displayName,

5
src/modules/Elsa.Python/Features/PythonFeature.cs
View file

@ -4,11 +4,13 @@ using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Python.Activities;
using Elsa.Python.Contracts;
using Elsa.Python.HostedServices;
using Elsa.Python.Options;
using Elsa.Python.Providers;
using Elsa.Python.Services;
using Elsa.Workflows;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Python.Features;
@ -52,5 +54,8 @@ public class PythonFeature : FeatureBase
// Activities.
Module.AddActivitiesFrom<PythonFeature>();
// UI property handlers.
Services.AddScoped<IPropertyUIHandler, RunPythonOptionsProvider>();
}
}

1
src/modules/Elsa.Workflows.Core/Contexts/WorkflowExecutionContext.cs
View file

@ -533,6 +533,7 @@ public partial class WorkflowExecutionContext : IExecutionContext
var parentContext = options?.Owner;
var parentExpressionExecutionContext = parentContext?.ExpressionExecutionContext ?? ExpressionExecutionContext;
var properties = ExpressionExecutionContextExtensions.CreateActivityExecutionContextPropertiesFrom(this, Input);
properties[ExpressionExecutionContextExtensions.ActivityKey] = activity;
var memory = new MemoryRegister();
var now = SystemClock.UtcNow;
var expressionExecutionContext = new ExpressionExecutionContext(ServiceProvider, memory, parentExpressionExecutionContext, properties, CancellationToken);

8
src/modules/Elsa.Workflows.Core/Contracts/IPayloadSerializer.cs
View file

@ -28,6 +28,14 @@ public interface IPayloadSerializer
/// <returns>The deserialized state.</returns>
object Deserialize(string serializedData);
/// <summary>
/// Deserializes the specified serialized state.
/// </summary>
/// <param name="serializedData">The serialized state.</param>
/// <param name="type">The type to deserialize the state into.</param>
/// <returns>The deserialized state.</returns>
object Deserialize(string serializedData, Type type);
/// <summary>
/// Deserializes the specified serialized state.
/// </summary>

10
src/modules/Elsa.Workflows.Core/Extensions/ExpressionExecutionContextExtensions.cs
View file

@ -37,6 +37,11 @@ public static class ExpressionExecutionContextExtensions
/// The key used to store the workflow in the <see cref="ExpressionExecutionContext.TransientProperties"/> dictionary.
/// </summary>
public static readonly object WorkflowKey = new();
/// <summary>
/// The key used to store the activity in the <see cref="ExpressionExecutionContext.TransientProperties"/> dictionary.
/// </summary>
public static readonly object ActivityKey = new();
/// <summary>
/// Creates a dictionary for the specified <see cref="WorkflowExecutionContext"/> and <see cref="ActivityExecutionContext"/>.
@ -78,6 +83,11 @@ public static class ExpressionExecutionContextExtensions
/// Returns the <see cref="ActivityExecutionContext"/> of the specified <see cref="ExpressionExecutionContext"/>
/// </summary>
public static bool TryGetActivityExecutionContext(this ExpressionExecutionContext context, out ActivityExecutionContext activityExecutionContext) => context.TransientProperties.TryGetValue(ActivityExecutionContextKey, out activityExecutionContext!);
/// <summary>
/// Returns the <see cref="Activity"/> of the specified <see cref="ExpressionExecutionContext"/>
/// </summary>
public static IActivity GetActivity(this ExpressionExecutionContext context) => (IActivity)context.TransientProperties[ActivityKey];
/// <summary>
/// Returns the value of the specified input.

38
src/modules/Elsa.Workflows.Core/Extensions/TriggerExtensions.cs
View file

@ -2,6 +2,8 @@ using Elsa.Expressions.Contracts;
using Elsa.Expressions.Models;
using Elsa.Workflows;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Models;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
// ReSharper disable once CheckNamespace
@ -22,39 +24,57 @@ public static class TriggerExtensions
var activityTypeName = TypeNameHelper.GenerateTypeName<T>();
return triggers.Where(x => x.Type == activityTypeName);
}
/// <summary>
/// Creates an expression execution context for the specified trigger.
/// </summary>
/// <param name="trigger">The trigger for which to create an expression execution context.</param>
/// <param name="activityDescriptor">The activity descriptor.</param>
/// <param name="serviceProvider">The service provider.</param>
/// <param name="context">The workflow indexing context.</param>
/// <param name="expressionEvaluator">The expression evaluator.</param>
/// <param name="logger">The logger.</param>
/// <returns>An expression execution context.</returns>
public static async Task<ExpressionExecutionContext> CreateExpressionExecutionContextAsync(
this ITrigger trigger,
IServiceProvider serviceProvider,
WorkflowIndexingContext context,
this ITrigger trigger,
ActivityDescriptor activityDescriptor,
IServiceProvider serviceProvider,
WorkflowIndexingContext context,
IExpressionEvaluator expressionEvaluator,
ILogger logger)
{
var inputs = trigger.GetInputs();
var assignedInputs = inputs.Where(x => x.MemoryBlockReference != null!).ToList();
var namedInputs = trigger.GetNamedInputs();
var assignedInputs = namedInputs.Where(x => x.Value.MemoryBlockReference != null!).ToList();
var register = context.GetOrCreateRegister(trigger);
var cancellationToken = context.CancellationToken;
var expressionInput = new Dictionary<string, object>();
var applicationProperties = ExpressionExecutionContextExtensions.CreateTriggerIndexingPropertiesFrom(context.Workflow, expressionInput);
applicationProperties[ExpressionExecutionContextExtensions.ActivityKey] = trigger;
var expressionExecutionContext = new ExpressionExecutionContext(serviceProvider, register, default, applicationProperties, cancellationToken);
// Evaluate activity inputs before requesting trigger data.
foreach (var input in assignedInputs)
foreach (var namedInput in assignedInputs)
{
var inputDescriptor = activityDescriptor.Inputs.FirstOrDefault(x => x.Name == namedInput.Key);
if (inputDescriptor == null)
{
logger.LogWarning("Input descriptor not found for input '{InputName}'", namedInput.Key);
continue;
}
if (!inputDescriptor.AutoEvaluate)
{
logger.LogDebug("Skipping input '{InputName}' because it is not set to auto-evaluate.", namedInput.Key);
continue;
}
var input = namedInput.Value;
var locationReference = input.MemoryBlockReference();
if(locationReference.Id == null!)
if (locationReference.Id == null!)
continue;
try
{
var value = await expressionEvaluator.EvaluateAsync(input, expressionExecutionContext);

8
src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs
View file

@ -1,6 +1,7 @@
using Elsa.Common;
using Elsa.Common.Features;
using Elsa.Common.Serialization;
using Elsa.CSharp.Activities;
using Elsa.Expressions.Features;
using Elsa.Extensions;
using Elsa.Features.Abstractions;
@ -230,10 +231,15 @@ public class WorkflowsFeature : FeatureBase
.AddScoped<IUIHintHandler, CheckListUIHintHandler>()
.AddScoped<IUIHintHandler, JsonEditorUIHintHandler>()
// UI property handlers.
.AddScoped<IPropertyUIHandler, StaticCheckListOptionsProvider>()
.AddScoped<IPropertyUIHandler, StaticDropDownOptionsProvider>()
.AddScoped<IPropertyUIHandler, JsonCodeOptionsProvider>()
// Logger state generators.
.AddSingleton(WorkflowLoggerStateGenerator)
.AddSingleton(ActivityLoggerStateGenerator)
// Log Persistence Strategies.
.AddScoped<ILogPersistenceStrategyService, DefaultLogPersistenceStrategyService>()
.AddScoped<ILogPersistenceStrategy, Include>()

2
src/modules/Elsa.Workflows.Core/Models/Bookmark.cs
View file

@ -23,7 +23,7 @@ public record Bookmark(
object? Payload,
string ActivityId,
string ActivityNodeId,
string ActivityInstanceId,
string? ActivityInstanceId,
DateTimeOffset CreatedAt,
bool AutoBurn = true,
string? CallbackMethodName = default,

2
src/modules/Elsa.Workflows.Core/Serialization/Converters/ExpandoObjectConverter.cs
View file

@ -12,7 +12,7 @@ public sealed class ExpandoObjectConverter : JsonConverter<object>
/// <inheritdoc />
public override void Write(Utf8JsonWriter writer, object value, JsonSerializerOptions options)
{
JsonSerializer.Serialize(writer, value, typeof(ExpandoObject), options);
JsonSerializer.Serialize(writer, value, value.GetType(), options);
}
/// <inheritdoc />

6
src/modules/Elsa.Workflows.Core/Serialization/Serializers/JsonPayloadSerializer.cs
View file

@ -41,6 +41,12 @@ public class JsonPayloadSerializer : IPayloadSerializer
return Deserialize<object>(payload);
}
public object Deserialize(string serializedData, Type type)
{
var options = GetOptions();
return JsonSerializer.Deserialize(serializedData, type, options)!;
}
/// <inheritdoc />
public object Deserialize(JsonElement payload)
{

89
src/modules/Elsa.Workflows.Core/Services/PropertyUIHandlerResolver.cs
View file

@ -2,48 +2,53 @@
using Elsa.Workflows.Attributes;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows;
namespace Elsa.Workflows;
/// <inheritdoc />
public class PropertyUIHandlerResolver(IServiceScopeFactory scopeFactory) : IPropertyUIHandlerResolver
{
public class PropertyUIHandlerResolver(IServiceScopeFactory scopeFactory) : IPropertyUIHandlerResolver
{
/// <inheritdoc />
public async ValueTask<IDictionary<string, object>> GetUIPropertiesAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken = default)
{
var inputAttribute = propertyInfo.GetCustomAttribute<InputAttribute>();
var result = new Dictionary<string, object>();
var uiHandlers = new List<Type>();
if (inputAttribute?.UIHandler != null)
uiHandlers.Add(inputAttribute.UIHandler);
if (inputAttribute?.UIHandlers != null)
uiHandlers.AddRange(inputAttribute.UIHandlers);
using var scope = scopeFactory.CreateScope();
var uiHintHandlers = scope.ServiceProvider.GetServices<IUIHintHandler>();
if (!string.IsNullOrWhiteSpace(inputAttribute?.UIHint))
{
var uiHintHandler = uiHintHandlers.FirstOrDefault(x => x.UIHint == inputAttribute.UIHint);
if (uiHintHandler != null)
{
var defaultHandlers = await uiHintHandler.GetPropertyUIHandlersAsync(propertyInfo, cancellationToken);
uiHandlers.AddRange(defaultHandlers);
}
}
foreach (var handlerType in uiHandlers)
{
var provider = (IPropertyUIHandler)ActivatorUtilities.GetServiceOrCreateInstance(scope.ServiceProvider, handlerType);
var properties = await provider.GetUIPropertiesAsync(propertyInfo, context, cancellationToken);
foreach (var property in properties)
result[property.Key] = property.Value;
}
return result;
}
public async ValueTask<IDictionary<string, object>> GetUIPropertiesAsync(PropertyInfo propertyInfo, object? context, CancellationToken cancellationToken = default)
{
var inputAttribute = propertyInfo.GetCustomAttribute<InputAttribute>();
var result = new Dictionary<string, object>();
var uiHandlers = new List<Type>();
if (inputAttribute?.UIHandler != null)
uiHandlers.Add(inputAttribute.UIHandler);
if (inputAttribute?.UIHandlers != null)
uiHandlers.AddRange(inputAttribute.UIHandlers);
using var scope = scopeFactory.CreateScope();
var uiHintHandlers = scope.ServiceProvider.GetServices<IUIHintHandler>();
if (!string.IsNullOrWhiteSpace(inputAttribute?.UIHint))
{
var uiHintHandler = uiHintHandlers.FirstOrDefault(x => x.UIHint == inputAttribute.UIHint);
if (uiHintHandler != null)
{
var defaultHandlers = await uiHintHandler.GetPropertyUIHandlersAsync(propertyInfo, cancellationToken);
uiHandlers.AddRange(defaultHandlers);
}
}
var propertyUIHandlers = scope.ServiceProvider.GetServices<IPropertyUIHandler>().ToList();
foreach (var handlerType in uiHandlers)
{
var provider = propertyUIHandlers.FirstOrDefault(x => x.GetType() == handlerType);
if (provider == null)
continue;
var properties = await provider.GetUIPropertiesAsync(propertyInfo, context, cancellationToken);
foreach (var property in properties)
result[property.Key] = property.Value;
}
return result;
}
}

28
src/modules/Elsa.Workflows.Management/Providers/DefaultExpressionDescriptorProvider.cs
View file

@ -53,19 +53,21 @@ public class DefaultExpressionDescriptorProvider : IExpressionDescriptorProvider
memoryBlockReferenceFactory: () => new Variable(),
deserialize: context =>
{
var valueElement = context.JsonElement.TryGetProperty("value", out var v) ? v : default;
var valueString = valueElement.GetValue()?.ToString();
try
{
var value = JsonSerializer.Deserialize(valueString, context.MemoryBlockType, context.Options);
return new Expression("Variable", value);
}
catch (Exception)
{
return new Expression("Variable", null);
}
var valueElement = context.JsonElement.TryGetProperty("value", out var v) ? v : default;
var valueString = valueElement.GetValue()?.ToString();
if (string.IsNullOrWhiteSpace(valueString))
return new Expression("Variable", null);
try
{
var value = JsonSerializer.Deserialize(valueString, context.MemoryBlockType, context.Options);
return new Expression("Variable", value);
}
catch (Exception)
{
return new Expression("Variable", null);
}
}
);
}

3
src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkResumer.cs
View file

@ -27,6 +27,9 @@ public interface IBookmarkResumer
/// Resumes the bookmark.
/// </summary>
Task<ResumeBookmarkResult> ResumeAsync<TActivity>(string bookmarkId, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default) where TActivity : IActivity;
/// Resumes the bookmark.
Task<ResumeBookmarkResult> ResumeAsync(ResumeBookmarkRequest request, CancellationToken cancellationToken = default);
/// <summary>
/// Resumes the bookmark.

6
src/modules/Elsa.Workflows.Runtime/Contracts/ITriggerInvoker.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Workflows.Runtime;
public interface ITriggerInvoker
{
Task<StartWorkflowResponse> InvokeAsync(InvokeTriggerRequest request, CancellationToken cancellationToken = default);
}

15
src/modules/Elsa.Workflows.Runtime/Entities/StoredBookmark.cs
View file

@ -1,4 +1,5 @@
using Elsa.Common.Entities;
using Elsa.Workflows.Models;
namespace Elsa.Workflows.Runtime.Entities;
@ -46,4 +47,18 @@ public class StoredBookmark : Entity
/// The date and time the bookmark was created.
/// </summary>
public DateTimeOffset CreatedAt { get; set; }
public Bookmark ToBookmark()
{
return new Bookmark
{
Id = Id,
Name = ActivityTypeName,
Hash = Hash,
CreatedAt = CreatedAt,
ActivityInstanceId = ActivityInstanceId,
Payload = Payload,
Metadata = Metadata
};
}
}

5
src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs
View file

@ -20,6 +20,7 @@ using Elsa.Workflows.Runtime.Providers;
using Elsa.Workflows.Runtime.Services;
using Elsa.Workflows.Runtime.Stores;
using Elsa.Workflows.Runtime.Tasks;
using Elsa.Workflows.Runtime.UIHints;
using Medallion.Threading;
using Medallion.Threading.FileSystem;
using Microsoft.Extensions.DependencyInjection;
@ -251,6 +252,7 @@ public class WorkflowRuntimeFeature : FeatureBase
.AddScoped<IBookmarksPersister, BookmarksPersister>()
.AddScoped<IBookmarkResumer, BookmarkResumer>()
.AddScoped<IBookmarkQueue, StoreBookmarkQueue>()
.AddScoped<ITriggerInvoker, TriggerInvoker>()
.AddScoped<IWorkflowCanceler, WorkflowCanceler>()
.AddScoped<IWorkflowCancellationService, WorkflowCancellationService>()
.AddScoped<IWorkflowActivationStrategyEvaluator, DefaultWorkflowActivationStrategyEvaluator>()
@ -297,6 +299,9 @@ public class WorkflowRuntimeFeature : FeatureBase
// Workflow definition providers.
.AddWorkflowDefinitionProvider<ClrWorkflowsProvider>()
// UI prooprty handlers.
.AddScoped<IPropertyUIHandler, DispatcherChannelOptionsProvider>()
// Domain handlers.
.AddCommandHandler<DispatchWorkflowCommandHandler>()

14
src/modules/Elsa.Workflows.Runtime/Requests/InvokeTriggerRequest.cs Normal file
View file

@ -0,0 +1,14 @@
using Elsa.Workflows.Activities;
namespace Elsa.Workflows.Runtime;
public class InvokeTriggerRequest
{
public Workflow Workflow { get; set; } = default!;
public string ActivityId { get; set; } = default!;
public string? CorrelationId { get; set; }
public IDictionary<string, object>? Input { get; set; }
public IDictionary<string, object>? Properties { get; set; }
public string? ParentWorkflowInstanceId { get; set; }
}

20
src/modules/Elsa.Workflows.Runtime/Requests/ResumeBookmarkRequest.cs Normal file
View file

@ -0,0 +1,20 @@
using Elsa.Workflows.Models;
namespace Elsa.Workflows.Runtime;
public class ResumeBookmarkRequest
{
public string WorkflowInstanceId { get; set; } = default!;
/// The ID of the bookmark that triggered the workflow instance, if any.
public string BookmarkId { get; set; } = default!;
/// The handle of the activity to schedule, if any.
public ActivityHandle? ActivityHandle { get; set; }
/// Any additional properties to associate with the workflow instance.
public IDictionary<string, object>? Properties { get; set; }
/// The input to the workflow instance, if any.
public IDictionary<string, object>? Input { get; set; }
}

16
src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs
View file

@ -55,6 +55,22 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
return await ResumeAsync(bookmarkFilter, options, cancellationToken);
}
public async Task<ResumeBookmarkResult> ResumeAsync(ResumeBookmarkRequest request, CancellationToken cancellationToken = default)
{
var runRequest = new RunWorkflowInstanceRequest
{
Input = request.Input,
Properties = request.Properties,
ActivityHandle = request.ActivityHandle,
BookmarkId = request.BookmarkId
};
var workflowInstanceId = request.WorkflowInstanceId;
var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, cancellationToken);
var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
return ResumeBookmarkResult.Found(response);
}
/// <inheritdoc />
public async Task<ResumeBookmarkResult> ResumeAsync(BookmarkFilter filter, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default)
{

44
src/modules/Elsa.Workflows.Runtime/Services/StimulusSender.cs
View file

@ -15,6 +15,7 @@ public class StimulusSender(
IBookmarkQueue bookmarkQueue,
IWorkflowRuntime workflowRuntime,
IWorkflowStarter workflowStarter,
ITriggerInvoker triggerInvoker,
ILogger<StimulusSender> logger) : IStimulusSender
{
/// <inheritdoc />
@ -56,24 +57,25 @@ public class StimulusSender(
foreach (var trigger in triggerBoundWorkflow.Triggers)
{
var startRequest = new StartWorkflowRequest
var triggerRequest = new InvokeTriggerRequest
{
Workflow = workflow,
CorrelationId = correlationId,
Workflow = workflow,
ActivityId = trigger.ActivityId,
Input = input,
Properties = properties,
ParentId = parentId,
TriggerActivityId = trigger.ActivityId
ParentWorkflowInstanceId = parentId
};
var startResponse = await workflowStarter.StartWorkflowAsync(startRequest, cancellationToken);
if (startResponse.CannotStart)
var response = await triggerInvoker.InvokeAsync(triggerRequest, cancellationToken);
if (response.CannotStart)
{
logger.LogWarning("Workflow activation strategy disallowed starting workflow {WorkflowDefinitionHandle} with correlation ID {CorrelationId}", workflow.DefinitionHandle, correlationId);
continue;
}
responses.Add(startResponse.ToRunWorkflowInstanceResponse());
responses.Add(response.ToRunWorkflowInstanceResponse());
}
}
@ -119,27 +121,21 @@ public class StimulusSender(
}
else
{
// If no bookmarks were matched, but bookmark-specific details were given, enqueue the request in case a matching bookmark is created in the near future.
// If no bookmarks were matched, enqueue the request in case a matching bookmark is created in the near future.
var workflowInstanceId = metadata?.WorkflowInstanceId;
var bookmarkId = metadata?.BookmarkId;
var activityInstanceId = metadata?.ActivityInstanceId;
var correlationId = metadata?.CorrelationId;
if (workflowInstanceId != null || bookmarkId != null || activityInstanceId != null || correlationId != null)
var bookmarkQueueItem = new NewBookmarkQueueItem
{
var bookmarkQueueItem = new NewBookmarkQueueItem
WorkflowInstanceId = workflowInstanceId,
BookmarkId = metadata?.BookmarkId,
StimulusHash = stimulusHash,
Options = new ResumeBookmarkOptions
{
WorkflowInstanceId = workflowInstanceId,
BookmarkId = metadata?.BookmarkId,
StimulusHash = stimulusHash,
Options = new ResumeBookmarkOptions
{
Input = input,
Properties = properties
}
};
await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken);
}
Input = input,
Properties = properties
}
};
await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken);
}
return responses;

16
src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs
View file

@ -23,6 +23,7 @@ public class TriggerIndexer : ITriggerIndexer
private readonly IExpressionEvaluator _expressionEvaluator;
private readonly IIdentityGenerator _identityGenerator;
private readonly ITriggerStore _triggerStore;
private readonly IActivityRegistry _activityRegistry;
private readonly INotificationSender _notificationSender;
private readonly IServiceProvider _serviceProvider;
private readonly IStimulusHasher _hasher;
@ -37,6 +38,7 @@ public class TriggerIndexer : ITriggerIndexer
IExpressionEvaluator expressionEvaluator,
IIdentityGenerator identityGenerator,
ITriggerStore triggerStore,
IActivityRegistry activityRegistry,
INotificationSender notificationSender,
IServiceProvider serviceProvider,
IStimulusHasher hasher,
@ -46,6 +48,7 @@ public class TriggerIndexer : ITriggerIndexer
_expressionEvaluator = expressionEvaluator;
_identityGenerator = identityGenerator;
_triggerStore = triggerStore;
_activityRegistry = activityRegistry;
_notificationSender = notificationSender;
_serviceProvider = serviceProvider;
_hasher = hasher;
@ -153,11 +156,18 @@ public class TriggerIndexer : ITriggerIndexer
{
var workflow = context.Workflow;
var cancellationToken = context.CancellationToken;
var expressionExecutionContext = await trigger.CreateExpressionExecutionContextAsync(_serviceProvider, context, _expressionEvaluator, _logger);
var triggerTypeName = trigger.Type;
var triggerDescriptor = _activityRegistry.Find(triggerTypeName, trigger.Version);
if (triggerDescriptor == null)
{
_logger.LogWarning("Could not find activity descriptor for activity type {ActivityType}", triggerTypeName);
return new List<StoredTrigger>(0);
}
var expressionExecutionContext = await trigger.CreateExpressionExecutionContextAsync(triggerDescriptor, _serviceProvider, context, _expressionEvaluator, _logger);
var triggerIndexingContext = new TriggerIndexingContext(context, expressionExecutionContext, trigger, cancellationToken);
var triggerData = await TryGetTriggerDataAsync(trigger, triggerIndexingContext);
var triggerTypeName = trigger.Type;
// If no trigger payloads were returned, create a null payload.
if (!triggerData.Any()) triggerData.Add(null!);

19
src/modules/Elsa.Workflows.Runtime/Services/TriggerInvoker.cs Normal file
View file

@ -0,0 +1,19 @@
namespace Elsa.Workflows.Runtime;
public class TriggerInvoker(IWorkflowStarter workflowStarter) : ITriggerInvoker
{
public async Task<StartWorkflowResponse> InvokeAsync(InvokeTriggerRequest request, CancellationToken cancellationToken = default)
{
var startRequest = new StartWorkflowRequest
{
Workflow = request.Workflow,
CorrelationId = request.CorrelationId,
Input = request.Input,
Properties = request.Properties,
ParentId = request.ParentWorkflowInstanceId,
TriggerActivityId = request.ActivityId,
};
return await workflowStarter.StartWorkflowAsync(startRequest, cancellationToken);
}
}