Add Elsa.Kafka Module for Kafka Integration with Message Sending and Receiving Activities

[sfmskywalker]

This PR introduces a new module, Elsa.Kafka, to support Kafka integration in Elsa Workflows. The module includes two key activities to enable Kafka-based messaging workflows:

• SendMessage: Sends a message to a Kafka topic using a specified producer configuration.
• MessageReceived: Triggers when a message is received on a specified topic or set of topics based on a given consumer configuration.

Key Features:

1. Message Correlation:

   • Allows specifying a predicate in the MessageReceived activity to filter messages.
   • The predicate has access to a message variable representing the received message.
   • Messages can be handled as raw strings or parsed into types like ExpandoObject or custom .NET classes.

2. Flexible Configuration:

   • Kafka producers and consumers can be defined via appsettings.json or custom implementations of IConsumerDefinitionProvider and IProducerDefinitionProvider.
   • Multiple implementations are supported to expand the set of available producers and consumers.

Configuration

An example appsettings.json configuration is provided to set up topics, producers, and consumers, specifying parameters such as bootstrap servers, topics to consume from, and correlation fields for message filtering:

{
   "Kafka": {
      "Topics": [
         { "Id": "topic-1", "Name": "topic-1" },
         { "Id": "topic-2", "Name": "topic-2" }
      ],
      "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" ]
         }
      ]
   }
}

Sample Workflows

Producer Workflow Example

A sample ProducerWorkflow using the SendMessage activity:

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" })
        };
    }
}

Consumer Workflow Example

A sample ConsumerWorkflow using the MessageReceived activity:

public class ConsumerWorkflow : WorkflowBase
{
    protected override void Build(IWorkflowBuilder builder)
    {
        var orderReceivedMessage = builder.WithVariable<ExpandoObject>();
        builder.Name = "Consumer Workflow";
        builder.Root = new Sequence
        {
            Activities =
            {
                new MessageReceived
                {
                    ConsumerDefinitionId = new("consumer-1"),
                    MessageType = new(typeof(ExpandoObject)),
Predicate = new(JavaScriptExpression.Create("message => message.OrderId == '1'")), 
                    Result = new(orderReceivedMessage),
                    CanStartWorkflow = true
                },
                new WriteLine(c => JsonSerializer.Serialize(orderReceivedMessage.Get(c)))
            }
        };
    }
}

The module supports parsing messages as ExpandoObject or custom .NET types, allowing for flexible message processing in workflows.

This addition provides essential Kafka messaging capabilities, expanding Elsa’s integration options with event-driven and message-based architectures.

---

This change is