Manipulando eventos externos em funções duráveis (Azure Functions)

Artigo
05/07/2024

As funções do Orchestrator têm a capacidade de esperar e ouvir eventos externos. Esta característica das Funções Duráveis é frequentemente útil para lidar com a interação humana ou outros gatilhos externos.

Nota

Os eventos externos são operações assíncronas unidirecionais. Eles não são adequados para situações em que o cliente que envia o evento precisa de uma resposta síncrona da função orchestrator.

Aguarde eventos

A API "wait-for-external-event" da ligação de gatilho de orquestração permite que uma função orchestrator aguarde e ouça de forma assíncrona um evento entregue por um cliente externo. A função orquestrador de escuta declara o nome do evento e a forma dos dados que espera receber.

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

Nota

O código C# anterior é para Durable Functions 2.x. Para Durable Functions 1.x, você deve usar DurableOrchestrationContext em vez de IDurableOrchestrationContext. Para obter mais informações sobre as diferenças entre versões, consulte o artigo Durable Functions versions .

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

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)

param($Context)

$approved = Start-DurableExternalEventListener -EventName "Approval"

if ($approved) {
    # approval granted - do the approved action
} else {
    # approval denied - send a notification
}

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

O exemplo anterior escuta um único evento específico e age quando ele é recebido.

Você pode ouvir vários eventos simultaneamente, como no exemplo a seguir, que aguarda uma das três notificações de eventos possíveis.

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

Nota

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

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)

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

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

O exemplo anterior escuta qualquer um dos vários eventos. Também é possível esperar por todos os 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);
}

Nota

O código anterior é para Durable Functions 2.x. Para Durable Functions 1.x, você deve usar DurableOrchestrationContext em vez de IDurableOrchestrationContext. Para obter mais informações sobre as diferenças entre versões, consulte o artigo Durable Functions versions .

No .NET, se a carga útil do evento não puder ser convertida no tipo Tesperado, uma exceção será lançada.

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

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)

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

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

A API "wait-for-external-event" aguarda indefinidamente por alguma entrada. O aplicativo de função pode ser descarregado com segurança enquanto espera. Se e quando um evento chega para essa instância de orquestração, ele é despertado automaticamente e processa imediatamente o evento.

Nota

Se o seu aplicativo de função usa o Plano de Consumo, nenhuma cobrança será cobrada enquanto uma função orquestradora estiver aguardando uma tarefa de evento externo, não importa quanto tempo ela espere.

Tal como acontece com as Funções de Atividade, os eventos externos têm uma garantia de entrega pelo menos uma vez . Isso significa que, sob certas condições (como reinicializações, dimensionamento, falhas, etc.), seu aplicativo pode receber duplicatas do mesmo evento externo. Portanto, recomendamos que os eventos externos contenham algum tipo de ID que permita que eles sejam manualmente eliminados da duplicação em orquestradores.

Enviar eventos

Você pode usar a API "raise-event" definida pela associação do cliente de orquestração para enviar um evento externo para uma orquestração. Você também pode usar a API HTTP de evento de aumento interna para enviar um evento externo para uma orquestração.

Um evento gerado inclui um ID de instância, um eventName e eventData como parâmetros. As funções do Orchestrator lidam com esses eventos usando as APIs "wait-for-external-event". O eventName deve corresponder nas extremidades de envio e recebimento para que o evento seja processado. Os dados do evento também devem ser serializáveis por JSON.

Internamente, os mecanismos "raise-event" enfileiram uma mensagem que é captada pela função de orquestrador em espera. Se a instância não estiver aguardando o nome do evento especificado , a mensagem de evento será adicionada a uma fila na memória. Se a instância de orquestração começar a ouvir esse nome de evento, ela verificará se há mensagens de evento na fila.

Nota

Se não houver nenhuma instância de orquestração com o ID de instância especificado, a mensagem de evento será descartada.

Abaixo está um exemplo de função acionada por fila que envia um evento "Approval" para uma instância de função orchestrator. O ID da instância de orquestração vem do corpo da mensagem de fila.

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

Nota

O código C# anterior é para Durable Functions 2.x. Para Durable Functions 1.x, você deve usar OrchestrationClient atributo DurableClient em vez do atributo e deve usar o DurableOrchestrationClient tipo de parâmetro em vez de IDurableOrchestrationClient. Para obter mais informações sobre as diferenças entre versões, consulte o artigo Durable Functions versions .

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);
    await client.raiseEvent(instanceId, "Approval", true);
};

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)

param($instanceId)

Send-DurableExternalEvent -InstanceId $InstanceId -EventName "Approval"

@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, a API "raise-event" enfileira uma mensagem que é captada pela função de orquestrador em espera. Se a instância não estiver aguardando o nome do evento especificado , a mensagem do evento será adicionada a um buffer na memória. Se a instância de orquestração começar a ouvir o nome desse evento, ela verificará se há mensagens de evento no buffer e acionará a tarefa que estava aguardando.

Nota

Se não houver nenhuma instância de orquestração com o ID de instância especificado, a mensagem de evento será descartada.

HTTP

A seguir está um exemplo de uma solicitação HTTP que gera um evento "Approval" para uma instância de orquestração.

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

"true"

Nesse caso, o ID da instância é codificado como MyInstanceId.

Próximos passos

Saiba como implementar o tratamento de erros

Executar uma amostra que aguarda a interação humana

Manipulando eventos externos em funções duráveis (Azure Functions)

Aguarde eventos

Enviar eventos

HTTP

Próximos passos

Recursos adicionais