YMJake.Flowable.ExternalWorker.Client 1.0.0

Flowable External Worker Client for .NET

A .NET client library for Flowable External Workers with ASP.NET Core integration and pipeline-based configuration.

Installation

dotnet add package Flowable.ExternalWorker.Client

Quick Start

Configuration

{
  "FlowableExternalWorker": {
    "FlowableHost": "http://localhost:8091/flowable-work",
    "BasicAuthUsername": "rest-admin",
    "BasicAuthPassword": "test",
    "DefaultConcurrency": 2,
    "DefaultPollingIntervalSeconds": 15,
    "DefaultNumberOfTasks": 3,
    "DefaultLockDuration": "PT10M"
  }
}

Setup

// Program.cs
builder.Services.AddFlowableExternalWorker(options =>
{
    builder.Configuration.GetSection("FlowableExternalWorker").Bind(options);
});

// Register handlers
builder.Services.AddScoped<OrderProcessingHandler>();

// Configure workers with pipeline
builder.Services.AddFlowableWorker("processOrder")
    .AddHandler<OrderProcessingHandler>()
    .WithConcurrency(3)
    .WithPollingInterval(TimeSpan.FromSeconds(10));

Worker Implementation

Basic Handler

public class OrderProcessingHandler : IWorkerHandler
{
    private readonly ILogger<OrderProcessingHandler> _logger;

    public OrderProcessingHandler(ILogger<OrderProcessingHandler> logger) => _logger = logger;

    public async Task HandleAsync(IWorkerPipelineContext context)
    {
        _logger.LogInformation("Processing order {OrderId}", context.Job.Id);
        
        var customerId = context.Job.Variables.GetValueOrDefault("customerId");
        // Your business logic here
        
        await Task.Delay(1000, context.CancellationToken);
        
        context.Result = context.ResultBuilder.Success()
            .Variable("processedAt", DateTime.UtcNow)
            .Variable("status", "completed");
        
        _logger.LogInformation("Order processed successfully");
    }
}

Handler with Error Handling

public class PaymentValidationHandler : IWorkerHandler
{
    private readonly ILogger<PaymentValidationHandler> _logger;

    public PaymentValidationHandler(ILogger<PaymentValidationHandler> logger) => _logger = logger;

    public async Task HandleAsync(IWorkerPipelineContext context)
    {
        try
        {
            var amount = Convert.ToDecimal(context.Job.Variables.GetValueOrDefault("amount", 0));
            
            if (amount <= 0)
            {
                context.Result = context.ResultBuilder.Failure()
                    .WithMessage("Invalid payment amount")
                    .WithDetails($"Amount must be > 0, was {amount}")
                    .WithRetries(0);
                return;
            }
            
            if (amount > 10000)
            {
                context.Result = context.ResultBuilder.BpmnError("AMOUNT_TOO_HIGH")
                    .Variable("maxAllowed", 10000);
                return;
            }
            
            // Simulate validation
            await Task.Delay(500, context.CancellationToken);
            
            context.Result = context.ResultBuilder.Success()
                .Variable("validatedAt", DateTime.UtcNow)
                .Variable("validatedAmount", amount);
        }
        catch (Exception ex)
        {
            context.Result = context.ResultBuilder.Failure().Error(ex);
        }
    }
}

// Configure with middleware (optional)
builder.Services.AddFlowableWorker("validatePayment")
    .AddHandler<PaymentValidationHandler>()
    .WithConcurrency(2)
    .ConfigurePipeline(pipeline =>
    {
        // Add simple logging middleware
        pipeline.Use(next => async context =>
        {
            var logger = context.ServiceProvider.GetRequiredService<ILogger<PaymentValidationHandler>>();
            logger.LogInformation("Processing {JobId}", context.Job.Id);
            
            await next(context);
            
            logger.LogInformation("Completed {JobId}", context.Job.Id);
        });
    });

Features

• 🎯 Simple worker configuration with fluent API
• 🔧 Native ASP.NET Core integration and dependency injection
• 💪 Strong typing with full IntelliSense support
• 📝 Easy variable access with extension methods
• ⚡ Concurrent worker processing with configurable concurrency
• 🔄 Built-in error handling and retry mechanisms
• 🎛️ Optional middleware pipeline for advanced scenarios
• ⚙️ Per-worker configuration options

License

Apache License 2.0