Обработка внешних событий в устойчивых функциях (Функции Azure)

Функции оркестратора могут ожидать передачи внешних событий. Эта возможность устойчивых функций часто используется для обработки действий человека или других внешних триггеров.

Примечание.

Внешние события это односторонние асинхронные операции. Они не подходят для ситуаций, где клиенту, отправляющему событие, требуется синхронный ответ функции оркестратора.

Ожидание событий

API-интерфейс wait-for-external-event привязки триггера оркестрации позволяет функции оркестратора асинхронно ожидать и ожидать события, доставленного внешним клиентом. Функция оркестратора, ожидающая передачи данных, объявляет имя события и форму данных, которую ожидает получить.

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

Примечание.

Предыдущий пример на C# предназначен для Устойчивых функций версии 2.x. Для расширения Устойчивые функции 1.x необходимо использовать DurableOrchestrationContext вместо IDurableOrchestrationContext. Дополнительные сведения о различиях между версиями см. в статье Версии устойчивых функций.

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

В предыдущем примере ожидается передача определенного одиночного события и предпринимается действие при его получении.

Можно одновременно прослушивать несколько событий, как показано в следующем примере, в котором ожидается одно из трех возможных уведомлений о событиях.

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

Примечание.

Предыдущий пример на C# предназначен для Устойчивых функций версии 2.x. Для расширения Устойчивые функции 1.x необходимо использовать DurableOrchestrationContext вместо IDurableOrchestrationContext. Дополнительные сведения о различиях между версиями см. в статье Версии устойчивых функций.

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

В предыдущем примере ожидается любое из нескольких событий. Также возможно реализовать ожидание всех событий.

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

Примечание.

Предыдущий пример предназначен для Устойчивых функций версии 2.x. Для расширения Устойчивые функции 1.x необходимо использовать DurableOrchestrationContext вместо IDurableOrchestrationContext. Дополнительные сведения о различиях между версиями см. в статье Версии устойчивых функций.

Если полезные данные события невозможно преобразовать в ожидаемый тип T, в .NET выдается исключение.

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

API wait-for-external-event ожидает ввод данных в течение неограниченного времени. Приложение-функцию можно безопасно выгрузить во время ожидания. Когда событие поступит в этот экземпляр оркестрации, он автоматически активируется и немедленно обработает событие.

Примечание.

Если в приложении-функции используется план потребления, когда функция оркестратора ожидает задачи внешнего события, плата не взимается, независимо от того, сколько длится ожидание.

Как и в случае с функциями действий, внешние события имеют по крайней мере однократную гарантию доставки. Это означает, что при определенных условиях (например, перезапуски, масштабирование, сбои и т. д.) приложение может получать дубликаты одного и того же внешнего события. Поэтому мы рекомендуем, чтобы внешние события содержали какой-то идентификатор, который позволяет вручную де дублироваться в оркестраторах.

Отправка событий

API raise-event, определенный привязкой клиента оркестрации, можно использовать для отправки внешнего события в оркестрацию. Чтобы отправить внешнее событие в оркестрацию, можно также использовать встроенный API вызова событий.

Вызванное событие включает идентификатор экземпляра, eventName и eventData в качестве параметров. Функции оркестратора обрабатывают эти события с помощью API "wait-for-external-event". Для того чтобы событие было обработано, значения eventName на отправляющей и принимающей сторонах должны совпадать. Данные событий также должны быть JSON-сериализуемыми.

Внутри системы механизмы "вызова событий" помещают в очередь сообщение, которое получает ожидающая функция оркестратора. Если экземпляр не ожидает события с указанным именем, сообщение о событии добавляется в очередь в памяти. Если экземпляр оркестрации позже начинает ожидать события с этим именем, он проверяет сообщения о событиях в очереди.

Примечание.

Если экземпляр оркестрации с указанным идентификатором экземпляра отсутствует, сообщение о событии отменяется.

Ниже приведен пример функции, активируемой с помощью очереди, которая отправляет событие "Approval" в экземпляр функции оркестратора. Идентификатор экземпляра оркестрации поступает из текста сообщения очереди.

C#

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

Примечание.

Предыдущий пример на C# предназначен для Устойчивых функций версии 2.x. Для расширения "Устойчивые функции" версии 1.x необходимо использовать атрибут OrchestrationClient, а не DurableClient. Также следует использовать тип параметра DurableOrchestrationClient вместо IDurableOrchestrationClient. Дополнительные сведения о различиях между версиями см. в статье Версии устойчивых функций.

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

Внутри системы API "raise-event" помещают в очередь сообщение, которое получает ожидающая функция оркестратора. Если экземпляр не ожидает события с указанным именем, сообщение о событии добавляется в буфер в памяти. Если экземпляр оркестрации позже начнет прослушивать это имя события, он проверит буфер сообщений о событиях и активирует задачу, ожидающую его.

Примечание.

Если экземпляр оркестрации с указанным идентификатором экземпляра отсутствует, сообщение о событии отменяется.

HTTP

Ниже приведен пример HTTP-запроса, который вызывает событие "Утверждение" в экземпляр оркестрации.

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

"true"

В данном случае идентификатор экземпляра жестко задан как MyInstanceId.

Следующие шаги

Узнайте, как реализовать обработку ошибок

Запуск примера с ожиданием действий человека