External Task Workers

External Task Workers ermöglichen die Verarbeitung von External Tasks aus BPMN-Prozessen. Der .NET Client bietet zwei Ansätze: manuelle Verarbeitung und ASP.NET Core Hosting Integration.

Übersicht

Manuelle Verarbeitung - Worker manuell verwalten und steuern
Hosting Integration - Integration in ASP.NET Core mit IHostedService

Was sind External Tasks?

External Tasks sind BPMN Service Tasks mit type="external" Attribut. Sie ermöglichen die asynchrone Verarbeitung von Aufgaben durch externe Worker.


<bpmn:serviceTask id="Task_ProcessPayment" name="Process Payment">
  <bpmn:extensionElements>
    <processcube:properties>
      <processcube:property name="module" value="ProcessPaymentWorker" />
    </processcube:properties>
  </bpmn:extensionElements>
</bpmn:serviceTask>

Schnellstart

Einfacher Worker


using ProcessCube.Engine;
using ProcessCube.Engine.ExternalTaskWorkers;
 
var engineAddress = "http://localhost:8000";
 
// Worker erstellen
var worker = ExternalTaskWorkerBuilder
    .Create(engineAddress)
    .WithTopic("ProcessPayment")
    .UseHandlerMethod(async (externalTask) =>
    {
        // Task-Payload auslesen
        var input = externalTask.Tokens[TokenType.OnEnter].GetPayload<PaymentInput>();
 
        Console.WriteLine($"Processing payment: {input.PaymentId}");
 
        // Geschäftslogik ausführen
        var result = await ProcessPaymentAsync(input);
 
        // Ergebnis zurückgeben
        return new PaymentResult(
            result.TransactionId,
            result.Status
        );
    })
    .Build();
 
// Worker starten
await worker.StartAsync();
 
Console.WriteLine("Worker gestartet. Drücke Enter zum Beenden...");
Console.ReadLine();
 
// Worker stoppen
await worker.StopAsync();
 
async Task<(string TransactionId, string Status)> ProcessPaymentAsync(PaymentInput input)
{
    // Payment-Logik hier
    await Task.Delay(1000);
    return (Guid.NewGuid().ToString(), "Success");
}
 
internal sealed record PaymentInput(string PaymentId, decimal Amount);
internal sealed record PaymentResult(string TransactionId, string Status);

External Task Workers prüfen kontinuierlich (Polling) die Engine auf neue Tasks und verarbeiten diese automatisch.

Worker-Lifecycle

Fetch: Worker fragt Engine nach verfügbaren External Tasks
Lock: Task wird für den Worker reserviert (Lock Duration)
Execute: Handler-Methode verarbeitet den Task
Complete: Ergebnis wird an Engine zurückgeliefert
Repeat: Worker fragt erneut nach Tasks

Worker-Settings


var worker = ExternalTaskWorkerBuilder
    .Create(engineAddress)
    .WithTopic("ProcessOrder")
    .WithSettings(settings =>
    {
        settings.MaxParallelWorkerCount = 5;
        settings.LockDuration = TimeSpan.FromMinutes(2);
        settings.LongPollingDuration = TimeSpan.FromSeconds(30);
        settings.IdleBetweenFetchesDuration = TimeSpan.FromSeconds(5);
    })
    .UseHandlerMethod(async (externalTask) =>
    {
        // Handler-Logik
        return new { status = "completed" };
    })
    .Build();

Worker-Settings Parameter

Parameter	Default	Beschreibung
`MaxParallelWorkerCount`	1	Anzahl paralleler Task-Verarbeitungen
`LockDuration`	100s	Zeit, die ein Task blockiert ist
`LongPollingDuration`	30s	Wartezeit beim Polling
`IdleBetweenFetchesDuration`	0s	Pause zwischen Polling-Zyklen
`HasSensitivePayload`	false	Unterbindet Logging von Payloads

Mehrere Topics verarbeiten


var paymentWorker = ExternalTaskWorkerBuilder
    .Create(engineAddress)
    .WithTopic("ProcessPayment")
    .UseHandlerMethod(HandlePaymentAsync)
    .Build();
 
var shippingWorker = ExternalTaskWorkerBuilder
    .Create(engineAddress)
    .WithTopic("ArrangeShipping")
    .UseHandlerMethod(HandleShippingAsync)
    .Build();
 
// Beide Workers starten
await Task.WhenAll(
    paymentWorker.StartAsync(),
    shippingWorker.StartAsync()
);

Fehlerbehandlung


var worker = ExternalTaskWorkerBuilder
    .Create(engineAddress)
    .WithTopic("ProcessOrder")
    .UseHandlerMethod(async (externalTask) =>
    {
        try
        {
            var input = externalTask.Tokens[TokenType.OnEnter].GetPayload<OrderInput>();
 
            // Verarbeitung
            var result = await ProcessOrderAsync(input);
 
            return new OrderResult(result.OrderId, "Completed");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Fehler bei Task-Verarbeitung: {ex.Message}");
 
            // Exception wird automatisch an Engine gemeldet
            // Task wird als fehlgeschlagen markiert
            throw;
        }
    })
    .Build();

Wenn der Handler eine Exception wirft, wird der External Task als fehlgeschlagen markiert und kann später wiederholt werden.

Mit Authentifizierung


using ProcessCube.Engine.Client.Identity;
 
var identityProvider = new ClientCredentialsIdentityProvider(
    authorityAddress: "http://localhost:11235",
    clientId: "worker-service",
    clientSecret: "secret-key",
    scope: "engine_etw"
);
 
var identity = await identityProvider.GetIdentityAsync();
 
var worker = ExternalTaskWorkerBuilder
    .Create(engineAddress, identity)
    .WithTopic("ProcessPayment")
    .UseHandlerMethod(HandlePaymentAsync)
    .Build();
 
await worker.StartAsync();

Nächste Schritte

Erkunden Sie die verschiedenen Ansätze zur Worker-Implementierung:

Manuelle Verarbeitung - Vollständige Kontrolle über Worker
Hosting Integration - Integration in ASP.NET Core