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

Funkce orchestrátoru 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-eventvazby triggeru orchestrace umožňuje funkci orchestrátoru asynchronně čekat a naslouchat události doručené externím klientem. Funkce naslouchacího orchestrátoru deklaruje název události a tvar dat , která očekává, že bude přijímat.

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 určený pro Durable Functions 2.x. Pro Durable Functions 1.x musíte místo IDurableOrchestrationContextpoužít DurableOrchestrationContext . Další informace o rozdílech mezi verzemi najdete v článku Durable Functions verze.

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

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

Můžete naslouchat více událostem současně, jako v následujícím příkladu, který čeká na jedno ze tří možných oznámení událostí.

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 určený pro Durable Functions 2.x. Pro Durable Functions 1.x musíte místo IDurableOrchestrationContextpoužít DurableOrchestrationContext . Další informace o rozdílech mezi verzemi najdete v článku Durable Functions verze.

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

Předchozí příklad naslouchá všem 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 musíte místo IDurableOrchestrationContextpoužít DurableOrchestrationContext . Další informace o rozdílech mezi verzemi najdete v článku Durable Functions verze.

Pokud v .NET datovou část události nelze 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. Aplikaci funkcí je možné při čekání bezpečně uvolnit. Pokud a když pro tuto instanci orchestrace dorazí událost, automaticky se probudí a událost okamžitě zpracuje.

Poznámka

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

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 také použít integrované rozhraní HTTP API pro vyvolání událostí .

Vyvolaná událost obsahuje ID instance, eventName a eventData jako parametry. Funkce orchestratoru tyto události zpracovávají pomocí rozhraní API wait-for-external-event . EventName se musí shodovat na straně odesílání i přijímání, aby se událost mohla zpracovat. Data události musí být také serializovatelná ve formátu JSON.

Mechanismy "raise-event" interně začtou do fronty zprávu, kterou vyzvedne čekající funkce orchestrátoru. Pokud instance nečeká na zadaný název události, zpráva o události se přidá do fronty v paměti. Pokud instance orchestrace později začne naslouchat názvu události, zkontroluje ve frontě zprávy 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 Approval instanci 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 určený pro Durable Functions 2.x. Pro Durable Functions 1.x musíte místo atributu DurableClient použít OrchestrationClient atribut a typ parametru DurableOrchestrationClientIDurableOrchestrationClientmísto . Další informace o rozdílech mezi verzemi najdete v článku Durable Functions verze.

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ě zařidí zprávu, která se vyzvedne čekající funkcí orchestrátoru. Pokud instance nečeká na zadaný název události, zpráva události se přidá do vyrovnávací paměti. Pokud instance orchestrace později začne naslouchat názvu události, zkontroluje ve vyrovnávací paměti 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 Approval pro instanci orchestration.

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