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

Funkcje programu Orchestrator mają możliwość oczekiwania i nasłuchiwania zdarzeń zewnętrznych. Ta funkcja rozszerzenia Durable Functions jest często przydatna do obsługi interakcji człowieka 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. Funkcja orkiestratora nasłuchiwania deklaruje nazwę zdarzenia i kształt danych , które oczekuje do odebrania.

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 rozszerzenia Durable Functions 2.x. W przypadku rozszerzenia Durable Functions 1.x należy użyć funkcji DurableOrchestrationContextIDurableOrchestrationContextzamiast . Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Wersje rozszerzenia 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
    }
}

Powyższy przykład nasłuchuje określonego pojedynczego zdarzenia i podejmuje działania po odebraniu.

Możesz nasłuchiwać wielu zdarzeń jednocześnie, na przykład w poniższym przykładzie, co 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 rozszerzenia Durable Functions 2.x. W przypadku rozszerzenia Durable Functions 1.x należy użyć funkcji DurableOrchestrationContextIDurableOrchestrationContextzamiast . Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Wersje rozszerzenia 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) {
        // ...
    }
}

Poprzedni przykład nasłuchuje dowolnego 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 rozszerzenia Durable Functions 2.x. W przypadku rozszerzenia Durable Functions 1.x należy użyć funkcji DurableOrchestrationContextIDurableOrchestrationContextzamiast . Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Wersje rozszerzenia 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 przez czas nieokreślony na niektóre dane wejściowe. Aplikacja funkcji może zostać bezpiecznie zwolniona podczas oczekiwania. Jeśli i po nadejściu zdarzenia dla tego wystąpienia orkiestracji, jest ono automatycznie wybudzone i natychmiast przetwarza zdarzenie.

Uwaga

Jeśli aplikacja funkcji korzysta z planu zużycia, opłaty za rozliczenia są naliczane, gdy funkcja orkiestratora oczekuje na zadanie zdarzenia zewnętrznego, bez względu na czas oczekiwania.

Podobnie jak w przypadku funkcji działania, zdarzenia zewnętrzne mają co najmniej jednokrotną gwarancję dostarczania. Oznacza to, że w pewnych warunkach (takich jak ponowne uruchomienia, skalowanie, awarie itp.), aplikacja może odbierać duplikaty tego samego zdarzenia zewnętrznego. W związku z tym zaleca się, aby zdarzenia zewnętrzne zawierały jakiś identyfikator, który umożliwia ich ręczne deduplikowanie w orkiestratorach.

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 HTTP zdarzenia w celu wysłania zdarzenia zewnętrznego do aranżacji.

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 w celu przetworzenia zdarzenia. Dane zdarzenia muszą również być serializowalne w formacie JSON.

Wewnętrznie mechanizmy "raise-event" w kolejce komunikat, który zostanie odebrany 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 później zacznie nasłuchiwać tej nazwy zdarzenia, sprawdzi kolejkę komunikatów o zdarzeniach.

Uwaga

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

Poniżej znajduje się przykładowa funkcja wyzwalana przez kolejkę, która wysyła zdarzenie "Zatwierdzenie" 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 rozszerzenia Durable Functions 2.x. W przypadku rozszerzenia 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ł Wersje rozszerzenia 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ącą orkiestratora. Jeśli wystąpienie nie oczekuje na określoną nazwę zdarzenia, komunikat zdarzenia zostanie dodany do buforu w pamięci. Jeśli wystąpienie orkiestracji później rozpocznie nasłuchiwanie tej nazwy zdarzenia, sprawdzi bufor komunikatów o zdarzeniach i wyzwoli zadanie, które czekało na to.

Uwaga

Jeśli nie ma wystąpienia orkiestracji z określonym identyfikatorem wystąpienia, komunikat zdarzenia 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 jako MyInstanceId.

Następne kroki

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

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