Obsługa zdarzeń zewnętrznych w Durable Functions (Azure Functions)

Funkcje programu Orchestrator mogą czekać i nasłuchiwać zdarzeń zewnętrznych. Ta funkcja Durable Functions jest często przydatna do obsługi interakcji z ludźmi lub innych wyzwalaczy zewnętrznych.

Uwaga

Zdarzenia zewnętrzne to jednokierunkowe operacje asynchroniczne. Nie są one odpowiednie w sytuacjach, w których klient wysyłający zdarzenie potrzebuje synchronicznej odpowiedzi z funkcji orkiestratora.

Czekaj na zdarzenia

Interfejs API "wait-for-external-event"powiązania wyzwalacza orkiestracji umożliwia asynchroniczne oczekiwanie i nasłuchiwanie zdarzenia dostarczonego przez klienta zewnętrznego przez funkcję orkiestratora. Funkcja orkiestratora nasłuchiwania deklaruje nazwę zdarzenia i kształt danych , które oczekuje.

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

Uwaga

Poprzedni kod języka C# jest przeznaczony dla Durable Functions 2.x. W przypadku Durable Functions 1.x należy użyć wartości DurableOrchestrationContext zamiast IDurableOrchestrationContext. Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Durable Functions versions (Wersje 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 = 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)

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

W poprzednim przykładzie nasłuchuje określonego pojedynczego zdarzenia i podejmuje działania po odebraniu.

Można nasłuchiwać wielu zdarzeń jednocześnie, jak w poniższym przykładzie, który czeka na jedno z trzech możliwych powiadomień o zdarzeniach.

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

Uwaga

Poprzedni kod języka C# jest przeznaczony dla Durable Functions 2.x. W przypadku Durable Functions 1.x należy użyć wartości DurableOrchestrationContext zamiast IDurableOrchestrationContext. Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Durable Functions versions (Wersje 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 = context.task_any([event1, event2, event3])
    if winner == event1:
        # ...
    elif winner == event2:
        # ...
    elif winner == event3:
        # ...

main = df.Orchestrator.create(orchestrator_function)

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

W poprzednim przykładzie nasłuchuje dowolnych z wielu zdarzeń. Istnieje również możliwość oczekiwania na wszystkie zdarzenia.

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

Uwaga

Poprzedni kod jest przeznaczony dla Durable Functions 2.x. W przypadku Durable Functions 1.x należy użyć wartości DurableOrchestrationContext zamiast IDurableOrchestrationContext. Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Durable Functions versions (Wersje Durable Functions).

Jeśli na platformie .NET nie można przekonwertować ładunku zdarzenia na oczekiwany typ T, zgłaszany jest wyjątek.

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)

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

Interfejs API "wait-for-external-event" czeka na pewne dane wejściowe w nieskończoność. Aplikacja funkcji może być bezpiecznie zwolniona podczas oczekiwania. Jeśli i po nadejściu zdarzenia dla tego wystąpienia orkiestracji, zostanie ono rozbudzone automatycznie i natychmiast przetworzy zdarzenie.

Uwaga

Jeśli aplikacja funkcji korzysta z planu zużycie, opłaty za rozliczenia są naliczane, gdy funkcja orkiestratora oczekuje na zadanie zdarzenia zewnętrznego, niezależnie od tego, jak długo czeka.

Wysyłanie zdarzeń

Możesz użyć interfejsu API "raise-event" zdefiniowanego przez powiązanie klienta orkiestracji , aby wysłać zdarzenie zewnętrzne do aranżacji. Możesz również użyć wbudowanego interfejsu API wywołania zdarzenia HTTP , aby wysłać zdarzenie zewnętrzne do orkiestracji.

Zgłoszone zdarzenie zawiera identyfikator wystąpienia, eventName i eventData jako parametry. Funkcje programu Orchestrator obsługują te zdarzenia przy użyciu interfejsów API "wait-for-external-event". Nazwa zdarzenia musi być zgodna zarówno z końcem wysyłania, jak i odbierania, aby zdarzenie zostało przetworzone. Dane zdarzenia muszą być również możliwe do serializacji w formacie JSON.

Wewnętrznie mechanizmy "raise-event" w kolejce komunikat, który jest odbierany przez funkcję oczekującego orkiestratora. Jeśli wystąpienie nie oczekuje na określoną nazwę zdarzenia, komunikat zdarzenia zostanie dodany do kolejki w pamięci. Jeśli wystąpienie orkiestracji rozpocznie nasłuchiwanie tej nazwy zdarzenia, sprawdzi kolejkę pod kątem komunikatów o zdarzeniach.

Uwaga

Jeśli nie ma wystąpienia orkiestracji z określonym identyfikatorem wystąpienia, komunikat o zdarzeniu zostanie odrzucony.

Poniżej znajduje się przykładowa funkcja wyzwalana w kolejce, która wysyła zdarzenie "Approval" do wystąpienia funkcji orkiestratora. Identyfikator wystąpienia orkiestracji pochodzi z treści komunikatu kolejki.

C#

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

Uwaga

Poprzedni kod języka C# jest przeznaczony dla Durable Functions 2.x. W przypadku Durable Functions 1.x należy użyć OrchestrationClient atrybutu zamiast atrybutu DurableClient i należy użyć typu parametru DurableOrchestrationClientIDurableOrchestrationClientzamiast . Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Durable Functions versions (Wersje 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)

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

Wewnętrznie interfejs API "raise-event" kolejkuje komunikat, który jest odbierany przez funkcję oczekującego orkiestratora. Jeśli wystąpienie nie oczekuje na określoną nazwę zdarzenia, komunikat zdarzenia zostanie dodany do bufora w pamięci. Jeśli wystąpienie orkiestracji rozpocznie nasłuchiwanie tej nazwy zdarzenia, sprawdzi bufor komunikatów o zdarzeniach i wyzwoli zadanie, które go czekało.

Uwaga

Jeśli nie ma wystąpienia orkiestracji z określonym identyfikatorem wystąpienia, komunikat o zdarzeniu zostanie odrzucony.

HTTP

Poniżej przedstawiono przykład żądania HTTP, które zgłasza zdarzenie "Zatwierdzenie" do wystąpienia aranżacji.

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

"true"

W takim przypadku identyfikator wystąpienia jest zakodowany na stałe jako MyInstanceId.

Następne kroki

Dowiedz się, jak zaimplementować obsługę błędów

Uruchamianie przykładu, który czeka na interakcję z człowiekiem