Control de eventos externos con Durable Functions (Azure Functions)

Artikulua
06/01/2023

Las funciones de orquestador tienen la capacidad de esperar y escuchar eventos externos. Esta característica de Durable Functions suele ser útil para controlar las interacciones humanas u otros desencadenadores externos.

Oharra

Los eventos externos son operaciones asincrónicas unidireccionales. No resultan adecuados para situaciones donde el cliente que envía el evento necesita una respuesta sincrónica de la función de orquestador.

Espera de eventos

La API "wait-for-external-event" del enlace del desencadenador de orquestación permite que una función del orquestador espere y escuche de forma asíncrona un evento entregado por un cliente externo. La función de orquestador de escucha declara el nombre del evento y la forma de los datos que espera recibir.

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

Oharra

El código de C# anterior corresponde a Durable Functions 2.x. En el caso de Durable Functions 1.x, debe usar DurableOrchestrationContext en lugar de IDurableOrchestrationContext. Para obtener más información sobre las diferencias entre versiones, vea el artículo Versiones de 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
    }
}

El ejemplo anterior escucha un evento único específico y toma medidas cuando se recibe.

Puede escuchar varios eventos al mismo tiempo, al igual que en el ejemplo siguiente, que espera una de tres notificaciones de eventos posibles.

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

Oharra

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

El ejemplo anterior escucha cualquiera de varios eventos posibles. También es posible esperar todos los eventos.

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

Oharra

El código anterior corresponde a Durable Functions 2.x. En el caso de Durable Functions 1.x, debe usar DurableOrchestrationContext en lugar de IDurableOrchestrationContext. Para obtener más información sobre las diferencias entre versiones, vea el artículo Versiones de Durable Functions.

En .NET, si la carga del evento no se puede convertir al tipo T esperado, se produce una excepción.

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

La API "wait-for-external-event" espera indefinidamente alguna entrada. La aplicación de función puede descargarse con seguridad mientras espera. En el momento en que un evento llega a esta instancia de orquestación, esta se activa automáticamente y procesa de inmediato el evento.

Oharra

Si la aplicación de funciones usa el plan de consumo, no habrá cargos de facturación mientras una función del orquestador esté esperando una tarea de evento externa, sin importar el tiempo que espere.

Al igual que con las Funciones de actividad, los eventos externos tienen una garantía de entrega de al menos una vez. Esto significa que, en determinadas condiciones (como reinicios, escalado, bloqueos, etc.), la aplicación puede recibir duplicados del mismo evento externo. Por lo tanto, se recomienda que los eventos externos contengan algún tipo de identificador que les permita desduplicarse manualmente en orquestadores.

Envío de eventos

Puedes usar la API "raise-event" definida por el enlace del cliente de orquestación para enviar un evento externo a una orquestación. También puede usar la API de HTTP de generación de evento integrada para enviar un evento externo a una orquestación.

Un evento generado incluye valores de instance ID, eventName y eventData como parámetros. Las funciones de Orchestrator controlan estos eventos con las API "wait-for-external-event". El valor de eventName debe coincidir con los extremos de envío y recepción para que se procese el evento. Los datos del evento también se deben poder serializar con JSON.

De manera interna, los mecanismos de "raise-event" ponen en cola un mensaje que recoge la función de orquestador en espera. Si la instancia no está esperando el nombre de evento especificado, el mensaje del evento se agrega a una cola en memoria. Si la instancia de orquestación inicia posteriormente la escucha de ese nombre de evento, se comprobará si hay mensajes de eventos en la cola.

Oharra

Si no hay ninguna instancia de orquestación con el identificador de instancia especificado, se descartará el mensaje del evento.

A continuación se muestra una función desencadenada por la cola de ejemplo que envía un evento de aprobación a una instancia de la función de orquestador. El identificador de la instancia de orquestación procede del cuerpo del mensaje de la cola.

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

Oharra

El código de C# anterior corresponde a Durable Functions 2.x. En el caso de Durable Functions 1.x, debe usar el atributo OrchestrationClient en lugar del atributo DurableClient, además de usar el tipo de parámetro DurableOrchestrationClient en lugar de IDurableOrchestrationClient. Para obtener más información sobre las diferencias entre versiones, vea el artículo Versiones de 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);
}

Internamente, la API "raise-event" pone en cola un mensaje que es recogido por la función del orquestador en espera. Si la instancia no espera el nombre de evento especificado, el mensaje de evento se agrega a un búfer en memoria. Si la instancia de orquestación comienza a escuchar ese nombre de evento más tarde, comprobará el búfer en busca de mensajes de evento y activará la tarea que estaba esperando.

Oharra

Si no hay ninguna instancia de orquestación con el identificador de instancia especificado, se descartará el mensaje del evento.

HTTP

A continuación, se muestra una solicitud HTTP que genera un evento de "Aprobación" para una instancia de orquestación.

HTTP

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

"true"

En este caso, el id. de la instancia está codificado de forma rígida como MyInstanceId.

Pasos siguientes

Obtenga información sobre cómo implementar el control de errores

Ejecutar un ejemplo que espera interacción humana

Partekatu honen bidez:

Control de eventos externos con Durable Functions (Azure Functions)

Espera de eventos

Envío de eventos

HTTP

Pasos siguientes

Oharrak

Baliabide gehigarriak