Zpracování externích událostí v Durable Functions (Azure Functions)

Funkce orchestratoru mají možnost čekat a naslouchat externím událostem. Tato funkce Durable Functions je často užitečná pro zpracování lidské interakce nebo jiných externích triggerů.

Poznámka:

Externí události jsou jednosměrné asynchronní operace. Nejsou vhodné pro situace, kdy klient odesílající událost potřebuje synchronní odpověď z funkce orchestrátoru.

Čekání na události

Rozhraní API "wait-for-external-event" vazby triggeru orchestrace umožňuje funkci orchestrátoru asynchronně čekat a naslouchat události doručované externím klientem. Funkce orchestrátoru naslouchá deklaruje název události a tvar dat , která očekává.

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
    }
}

Poznámka:

Předchozí kód jazyka C# je pro Durable Functions 2.x. Pro Durable Functions 1.x je nutné použít DurableOrchestrationContext místo IDurableOrchestrationContext. Další informace o rozdílech mezi verzemi najdete v článku o verzích Durable Functions.

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 = yield 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
    }
}

Předchozí příklad naslouchá konkrétní jedné události a provede akci při přijetí.

Můžete naslouchat více událostem souběžně, například v následujícím příkladu, což čeká na jedno ze tří možných oznámení o událostech.

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)
    {
        // ...
    }
}

Poznámka:

Předchozí kód jazyka C# je pro Durable Functions 2.x. Pro Durable Functions 1.x je nutné použít DurableOrchestrationContext místo IDurableOrchestrationContext. Další informace o rozdílech mezi verzemi najdete v článku o verzích Durable Functions.

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 = yield 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) {
        // ...
    }
}

Předchozí příklad naslouchá libovolné z více událostí. Je také možné počkat na všechny události.

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

Poznámka:

Předchozí kód je pro Durable Functions 2.x. Pro Durable Functions 1.x je nutné použít DurableOrchestrationContext místo IDurableOrchestrationContext. Další informace o rozdílech mezi verzemi najdete v článku o verzích Durable Functions.

Pokud v .NET nelze datovou část události převést na očekávaný typ T, vyvolá se výjimka.

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

Rozhraní API "wait-for-external-event" čeká na určitý vstup neomezeně dlouho. Aplikace funkcí může být při čekání bezpečně uvolněna. Pokud a když událost dorazí pro tuto instanci orchestrace, probouzí se automaticky a okamžitě zpracuje událost.

Poznámka:

Pokud vaše aplikace funkcí používá plán Consumption, nebudou účtovány žádné poplatky za fakturaci, když funkce orchestrátoru čeká na úkol externí události bez ohledu na to, jak dlouho čeká.

Stejně jako u funkcí aktivit mají externí události záruku doručení alespoň jednou . To znamená, že za určitých podmínek (například restartování, škálování, chybové ukončení atd.) může vaše aplikace obdržet duplicity stejné externí události. Proto doporučujeme, aby externí události obsahovaly určitý druh ID, který umožňuje jejich ruční odstranění v orchestrátorech.

Odesílání událostí

K odeslání externí události do orchestrace můžete použít rozhraní API "raise-event" definované vazbou klienta orchestrace. K odeslání externí události do orchestrace můžete použít také integrované rozhraní API HTTP pro vyvolání události.

Vyvoláná událost obsahuje ID instance, eventName a eventData jako parametry. Funkce orchestratoru zpracovávají tyto události pomocí rozhraní API "wait-for-external-event" . Název události se musí shodovat jak na odesílajícím, tak i přijímajícím konci, aby se událost zpracovávala. Data události musí být také serializovatelná ve formátu JSON.

Mechanismy "raise-event" interně vytvoří frontu zprávy, která se vyzvedne funkcí čekající orchestrátoru. Pokud instance nečeká na zadaný název události, přidá se zpráva události do fronty v paměti. Pokud instance orchestrace později začne naslouchat názvu události, zkontroluje frontu zpráv událostí.

Poznámka:

Pokud neexistuje žádná instance orchestrace se zadaným ID instance, zpráva události se zahodí.

Níže je příklad funkce aktivované frontou, která odesílá událost Schválení do instance funkce orchestrátoru. ID instance orchestrace pochází z textu zprávy fronty.

C#

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

Poznámka:

Předchozí kód jazyka C# je pro Durable Functions 2.x. Pro Durable Functions 1.x, musíte použít OrchestrationClient atribut místo DurableClient atributu a musíte použít DurableOrchestrationClient typ parametru místo IDurableOrchestrationClient. Další informace o rozdílech mezi verzemi najdete v článku o verzích Durable Functions.

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

Rozhraní API "raise-event" interně vypíše zprávu, která se vyzvedne funkcí čekající orchestrátoru. Pokud instance nečeká na zadaný název události, zpráva události se přidá do vyrovnávací paměti v paměti. Pokud instance orchestrace později začne naslouchat názvu události, zkontroluje vyrovnávací paměť pro zprávy událostí a aktivuje úlohu, která na ni čekala.

Poznámka:

Pokud neexistuje žádná instance orchestrace se zadaným ID instance, zpráva události se zahodí.

HTTP

Následuje příklad požadavku HTTP, který vyvolá událost Schválení instanci orchestrace.

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

"true"

V tomto případě je ID instance pevně zakódované jako MyInstanceId.

Další kroky

Informace o implementaci zpracování chyb

Spuštění ukázky, která čeká na lidskou interakci