Externe gebeurtenissen verwerken in Durable Functions (Azure Functions)

Orchestrator-functies hebben de mogelijkheid om te wachten en te luisteren naar externe gebeurtenissen. Deze functie van Durable Functions is vaak handig voor het verwerken van menselijke interactie of andere externe triggers.

Notitie

Externe gebeurtenissen zijn asynchrone bewerkingen in één richting. Ze zijn niet geschikt voor situaties waarin de client die de gebeurtenis verzendt, een synchrone reactie van de orchestratorfunctie nodig heeft.

Wachten op gebeurtenissen

Met de API 'wait-for-external-event' van de orchestration-triggerbinding kan een orchestrator-functie asynchroon wachten en luisteren naar een gebeurtenis die door een externe client wordt geleverd. De listening orchestrator-functie declareert de naam van de gebeurtenis en de vorm van de gegevens die worden verwacht te ontvangen.

C#

[FunctionName("BudgetApproval")]
public static async Task Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    bool approved = await context.WaitForExternalEvent<bool>("Approval");
    if (approved)
    {
        // approval granted - do the approved action
    }
    else
    {
        // approval denied - send a notification
    }
}

Notitie

De vorige C#-code is voor Durable Functions 2.x. Voor Durable Functions 1.x moet u in DurableOrchestrationContext plaats van IDurableOrchestrationContext. Zie het artikel Over Durable Functions-versies voor meer informatie over de verschillen tussen versies.

JavaScript

const df = require("durable-functions");

module.exports = df.orchestrator(function*(context) {
    const approved = yield context.df.waitForExternalEvent("Approval");
    if (approved) {
        // approval granted - do the approved action
    } else {
        // approval denied - send a notification
    }
});

Python

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    approved = context.wait_for_external_event('Approval')
    if approved:
        # approval granted - do the approved action
    else:
        # approval denied - send a notification

main = df.Orchestrator.create(orchestrator_function)

Powershell

param($Context)

$approved = Start-DurableExternalEventListener -EventName "Approval"

if ($approved) {
    # approval granted - do the approved action
} else {
    # approval denied - send a notification
}

Java

@FunctionName("WaitForExternalEvent")
public void waitForExternalEvent(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    boolean approved = ctx.waitForExternalEvent("Approval", boolean.class).await();
    if (approved) {
        // approval granted - do the approved action
    } else {
        // approval denied - send a notification
    }
}

In het voorgaande voorbeeld wordt geluisterd naar een specifieke gebeurtenis en wordt actie ondernomen wanneer deze wordt ontvangen.

U kunt gelijktijdig naar meerdere gebeurtenissen luisteren, zoals in het volgende voorbeeld, waarbij wordt gewacht op een van de drie mogelijke gebeurtenismeldingen.

C#

[FunctionName("Select")]
public static async Task Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var event1 = context.WaitForExternalEvent<float>("Event1");
    var event2 = context.WaitForExternalEvent<bool>("Event2");
    var event3 = context.WaitForExternalEvent<int>("Event3");

    var winner = await Task.WhenAny(event1, event2, event3);
    if (winner == event1)
    {
        // ...
    }
    else if (winner == event2)
    {
        // ...
    }
    else if (winner == event3)
    {
        // ...
    }
}

Notitie

De vorige C#-code is voor Durable Functions 2.x. Voor Durable Functions 1.x moet u in DurableOrchestrationContext plaats van IDurableOrchestrationContext. Zie het artikel Over Durable Functions-versies voor meer informatie over de verschillen tussen versies.

JavaScript

const df = require("durable-functions");

module.exports = df.orchestrator(function*(context) {
    const event1 = context.df.waitForExternalEvent("Event1");
    const event2 = context.df.waitForExternalEvent("Event2");
    const event3 = context.df.waitForExternalEvent("Event3");

    const winner = yield context.df.Task.any([event1, event2, event3]);
    if (winner === event1) {
        // ...
    } else if (winner === event2) {
        // ...
    } else if (winner === event3) {
        // ...
    }
});

Python

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    event1 = context.wait_for_external_event('Event1')
    event2 = context.wait_for_external_event('Event2')
    event3 = context.wait_for_external_event('Event3')

    winner = context.task_any([event1, event2, event3])
    if winner == event1:
        # ...
    elif winner == event2:
        # ...
    elif winner == event3:
        # ...

main = df.Orchestrator.create(orchestrator_function)

Powershell

param($Context)

$event1 = Start-DurableExternalEventListener -EventName "Event1" -NoWait
$event2 = Start-DurableExternalEventListener -EventName "Event2" -NoWait
$event3 = Start-DurableExternalEventListener -EventName "Event3" -NoWait

$winner = Wait-DurableTask -Task @($event1, $event2, $event3) -Any

if ($winner -eq $event1) {
    # ...
} else if ($winner -eq $event2) {
    # ...
} else if ($winner -eq $event3) {
    # ...
}

Java

@FunctionName("Select")
public void selectOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    Task<Void> event1 = ctx.waitForExternalEvent("Event1");
    Task<Void> event2 = ctx.waitForExternalEvent("Event2");
    Task<Void> event3 = ctx.waitForExternalEvent("Event3");

    Task<?> winner = ctx.anyOf(event1, event2, event3).await();
    if (winner == event1) {
        // ...
    } else if (winner == event2) {
        // ...
    } else if (winner == event3) {
        // ...
    }
}

In het vorige voorbeeld wordt geluisterd naar een van de meerdere gebeurtenissen. Het is ook mogelijk om te wachten op alle gebeurtenissen.

C#

[FunctionName("NewBuildingPermit")]
public static async Task Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string applicationId = context.GetInput<string>();

    var gate1 = context.WaitForExternalEvent("CityPlanningApproval");
    var gate2 = context.WaitForExternalEvent("FireDeptApproval");
    var gate3 = context.WaitForExternalEvent("BuildingDeptApproval");

    // all three departments must grant approval before a permit can be issued
    await Task.WhenAll(gate1, gate2, gate3);

    await context.CallActivityAsync("IssueBuildingPermit", applicationId);
}

Notitie

De vorige code is bedoeld voor Durable Functions 2.x. Voor Durable Functions 1.x moet u in DurableOrchestrationContext plaats van IDurableOrchestrationContext. Zie het artikel Over Durable Functions-versies voor meer informatie over de verschillen tussen versies.

Als in .NET de nettolading van de gebeurtenis niet kan worden geconverteerd naar het verwachte type T, wordt er een uitzondering gegenereerd.

JavaScript

const df = require("durable-functions");

module.exports = df.orchestrator(function*(context) {
    const applicationId = context.df.getInput();

    const gate1 = context.df.waitForExternalEvent("CityPlanningApproval");
    const gate2 = context.df.waitForExternalEvent("FireDeptApproval");
    const gate3 = context.df.waitForExternalEvent("BuildingDeptApproval");

    // all three departments must grant approval before a permit can be issued
    yield context.df.Task.all([gate1, gate2, gate3]);

    yield context.df.callActivity("IssueBuildingPermit", applicationId);
});

Python

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    application_id = context.get_input()
    
    gate1 = context.wait_for_external_event('CityPlanningApproval')
    gate2 = context.wait_for_external_event('FireDeptApproval')
    gate3 = context.wait_for_external_event('BuildingDeptApproval')

    yield context.task_all([gate1, gate2, gate3])
    yield context.call_activity('IssueBuildingPermit', application_id)

main = df.Orchestrator.create(orchestrator_function)

Powershell

param($Context)

$applicationId = $Context.Input
$gate1 = Start-DurableExternalEventListener -EventName "CityPlanningApproval" -NoWait
$gate2 = Start-DurableExternalEventListener -EventName "FireDeptApproval" -NoWait
$gate3 = Start-DurableExternalEventListener -EventName "BuildingDeptApproval" -NoWait

Wait-DurableTask -Task @($gate1, $gate2, $gate3)

Invoke-ActivityFunction -FunctionName 'IssueBuildingPermit' -Input $applicationId

Java

@FunctionName("NewBuildingPermit")
public void newBuildingPermit(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String applicationId = ctx.getInput(String.class);

    Task<Void> gate1 = ctx.waitForExternalEvent("CityPlanningApproval");
    Task<Void> gate2 = ctx.waitForExternalEvent("FireDeptApproval");
    Task<Void> gate3 = ctx.waitForExternalEvent("BuildingDeptApproval");

    // all three departments must grant approval before a permit can be issued
    ctx.allOf(List.of(gate1, gate2, gate3)).await();

    ctx.callActivity("IssueBuildingPermit", applicationId).await();
}

De API 'wait-for-external-event' wacht voor onbepaalde tijd op enige invoer. De functie-app kan veilig worden verwijderd tijdens het wachten. Als en wanneer een gebeurtenis binnenkomt voor dit indelingsexemplaar, wordt deze automatisch geactiveerd en onmiddellijk de gebeurtenis verwerkt.

Notitie

Als uw functie-app gebruikmaakt van het verbruiksabonnement, worden er geen factureringskosten in rekening gebracht terwijl een orchestrator-functie wacht op een externe gebeurtenistaak, ongeacht hoe lang deze wacht.

Net als bij Activiteitsfuncties hebben externe gebeurtenissen een garantie voor ten minste één levering . Dit betekent dat uw toepassing onder bepaalde voorwaarden (zoals opnieuw opstarten, schalen, crashes, enzovoort) dubbele waarden van dezelfde externe gebeurtenis kan ontvangen. Daarom raden we aan dat externe gebeurtenissen een soort id bevatten waarmee ze handmatig kunnen worden gededupliceerd in orchestrators.

Gebeurtenissen verzenden

U kunt de API 'raise-event' gebruiken die is gedefinieerd door de orchestration-clientbinding om een externe gebeurtenis naar een indeling te verzenden. U kunt ook de ingebouwde HTTP-API voor het genereren van gebeurtenissen gebruiken om een externe gebeurtenis naar een indeling te verzenden.

Een gegenereerde gebeurtenis bevat een exemplaar-id, een eventName en eventData als parameters. Orchestrator-functies verwerken deze gebeurtenissen met behulp van de API's 'wait-for-external-event '. De eventName moet overeenkomen op zowel het verzenden als ontvangen, zodat de gebeurtenis kan worden verwerkt. De gebeurtenisgegevens moeten ook JSON-serializeerbaar zijn.

Intern worden met de mechanismen 'raise-event' een bericht weergegeven dat wordt opgehaald door de wachtende orchestratorfunctie. Als het exemplaar niet wacht op de opgegeven gebeurtenisnaam, wordt het gebeurtenisbericht toegevoegd aan een wachtrij in het geheugen. Als het indelingsexemplaar later begint te luisteren naar die gebeurtenisnaam, wordt de wachtrij gecontroleerd op gebeurtenisberichten.

Notitie

Als er geen indelingsexemplaar is met de opgegeven exemplaar-id, wordt het gebeurtenisbericht verwijderd.

Hieronder ziet u een voorbeeld van een door wachtrij geactiveerde functie waarmee een gebeurtenis 'Goedkeuring' naar een orchestratorfunctie-exemplaar wordt verzonden. De instantie-id van de indeling komt uit de hoofdtekst van het wachtrijbericht.

C#

[FunctionName("ApprovalQueueProcessor")]
public static async Task Run(
    [QueueTrigger("approval-queue")] string instanceId,
    [DurableClient] IDurableOrchestrationClient client)
{
    await client.RaiseEventAsync(instanceId, "Approval", true);
}

Notitie

De vorige C#-code is voor Durable Functions 2.x. Voor Durable Functions 1.x moet u het kenmerk gebruiken OrchestrationClient in plaats van het DurableClient kenmerk en moet u het DurableOrchestrationClient parametertype gebruiken in plaats van IDurableOrchestrationClient. Zie het artikel Over Durable Functions-versies voor meer informatie over de verschillen tussen versies.

JavaScript

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);
    await client.raiseEvent(instanceId, "Approval", true);
};

Python

import azure.functions as func
import azure.durable_functions as df

async def main(instance_id:str, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    await client.raise_event(instance_id, 'Approval', True)

Powershell

param($instanceId)

Send-DurableExternalEvent -InstanceId $InstanceId -EventName "Approval"

Java

@FunctionName("ApprovalQueueProcessor")
public void approvalQueueProcessor(
        @QueueTrigger(name = "instanceID", queueName = "approval-queue") String instanceID,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    durableContext.getClient().raiseEvent(instanceID, "Approval", true);
}

Intern wordt met de API 'raise-event' een bericht geïnventueerd dat wordt opgehaald door de wachtororfunctie. Als het exemplaar niet wacht op de opgegeven gebeurtenisnaam, wordt het gebeurtenisbericht toegevoegd aan een buffer in het geheugen. Als het indelingsexemplaar later begint met luisteren naar die gebeurtenisnaam, wordt de buffer gecontroleerd op gebeurtenisberichten en wordt de taak geactiveerd die erop wacht.

Notitie

Als er geen indelingsexemplaar is met de opgegeven exemplaar-id, wordt het gebeurtenisbericht verwijderd.

HTTP

Hier volgt een voorbeeld van een HTTP-aanvraag die een gebeurtenis 'Goedkeuring' genereert voor een indelingsinstantie.

POST /runtime/webhooks/durabletask/instances/MyInstanceId/raiseEvent/Approval&code=XXX
Content-Type: application/json

"true"

In dit geval wordt de exemplaar-id vastgelegd als MyInstanceId.

Volgende stappen

Meer informatie over het implementeren van foutafhandeling

Een voorbeeld uitvoeren dat wacht op menselijke interactie