Menangani kejadian eksternal di Durable Functions (Azure Functions)

Fungsi orkestrator memiliki kemampuan untuk menunggu dan mendengarkan kejadian eksternal. Fitur Durable Functions ini sering berguna untuk menangani interaksi manusia atau pemicu eksternal lainnya.

Catatan

Kejadian eksternal adalah operasi asinkron satu arah. Kejadian ini tidak cocok untuk situasi ketika klien yang mengirim kejadian membutuhkan respons sinkron dari fungsi orkestrator.

Menunggu kejadian

API "wait-for-external-event" dari pengikatan pemicu orkestrasi memungkinkan fungsi orkestrator menunggu dan mendengarkan peristiwa yang dikirimkan oleh klien eksternal secara asinkron. Fungsi orkestrator mendengarkan mendeklarasikan nama kejadian dan bentuk data yang diharapkan untuk diterima.

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

Catatan

Kode C# sebelumnya ditujukan untuk Durable Functions 2.x. Untuk Durable Functions 1.x, Anda harus menggunakan DurableOrchestrationContext dan bukan IDurableOrchestrationContext. Untuk informasi selengkapnya tentang perbedaan antara versi, lihat artikel Versi 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)

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

Contoh sebelumnya mendengarkan kejadian tunggal tertentu dan mengambil tindakan saat diterima.

Anda dapat mendengarkan beberapa kejadian secara bersamaan, seperti dalam contoh berikut, yang menunggu salah satu dari tiga kemungkinan pemberitahuan kejadian.

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

Catatan

Kode C# sebelumnya ditujukan untuk Durable Functions 2.x. Untuk Durable Functions 1.x, Anda harus menggunakan DurableOrchestrationContext dan bukan IDurableOrchestrationContext. Untuk informasi selengkapnya tentang perbedaan antara versi, lihat artikel Versi 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)

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

Contoh sebelumnya mendengarkan salah satu dari beberapa kejadian. Dimungkinkan juga untuk menunggu semua acara.

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

Catatan

Kode sebelumnya adalah untuk Durable Functions 2.x. Untuk Durable Functions 1.x, Anda harus menggunakan DurableOrchestrationContext dan bukan IDurableOrchestrationContext. Untuk informasi selengkapnya tentang perbedaan antara versi, lihat artikel Versi Durable Functions.

Dalam .NET, jika payload kejadian tidak dapat dikonversi ke jenis T yang diharapkan, pengecualian dilemparkan.

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" menunggu tanpa batas waktu untuk beberapa input. Aplikasi fungsi dapat dibongkar dengan aman saat menunggu. Jika dan ketika kejadian tiba untuk instans orkestrasi ini, ini akan terbangun secara otomatis dan segera memproses kejadian.

Catatan

Jika aplikasi fungsi Anda menggunakan Paket Konsumsi, tidak ada biaya penagihan yang dikenakan saat fungsi orkestrator sedang menunggu tugas peristiwa eksternal, tidak peduli berapa lama menunggu.

Seperti halnya Fungsi Aktivitas, peristiwa eksternal memiliki jaminan pengiriman setidaknya sekali . Ini berarti bahwa, dalam kondisi tertentu (seperti menghidupkan ulang, penskalaan, crash, dll.), aplikasi Anda mungkin menerima duplikat dari peristiwa eksternal yang sama. Oleh karena itu, kami menyarankan agar peristiwa eksternal berisi semacam ID yang memungkinkannya diduplikasi secara manual di orkestrator.

Mengirim aktivitas

Anda dapat menggunakan API "raise-event" yang ditentukan oleh pengikatan klien orkestrasi untuk mengirim peristiwa eksternal ke orkestrasi. Anda juga dapat menggunakan API HTTP tingkatkan kejadian bawaan untuk mengirim kejadian eksternal ke orkestrasi.

Kejadian yang ditingkatkan mencakup ID instans, eventName, dan eventData sebagai parameter. Fungsi orkestrator menangani peristiwa ini menggunakan API "wait-for-external-event". eventName harus cocok pada ujung pengiriman dan penerimaan agar acara dapat diproses. Data kejadian juga harus dapat diserialisasikan oleh JSON.

Secara internal, mekanisme "raise-event" mengantrekan pesan yang diambil oleh fungsi orkestrator yang menunggu. Jika instans tidak menunggu di nama kejadian yang ditentukan, pesan kejadian ditambahkan ke antrean dalam memori. Jika instans orkestrasi selanjutnya mulai mendengarkan nama kejadian tersebut, ini akan memeriksa antrean untuk pesan kejadian.

Catatan

Jika tidak ada instans orkestrasi dengan ID instans yang ditentukan, pesan kejadian akan dibuang.

Di bawah ini adalah contoh fungsi yang dipicu antrean yang mengirimkan kejadian "Persetujuan" ke instans fungsi orkestrator. ID instans orkestrasi berasal dari isi pesan antrean.

C#

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

Catatan

Kode C# sebelumnya ditujukan untuk Durable Functions 2.x. Untuk Durable Functions 1.x, Anda harus menggunakan atribut OrchestrationClient, bukan atribut DurableClient, dan jenis parameter DurableOrchestrationClient, bukan IDurableOrchestrationClient. Untuk informasi selengkapnya tentang perbedaan antara versi, lihat artikel Versi 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);
}

Secara internal, API "raise-event" mengantrekan pesan yang diambil oleh fungsi orkestrator yang menunggu. Jika instans tidak menunggu nama kejadian, yang ditentukan, pesan peristiwa ditambahkan ke buffer dalam memori. Jika instans orkestrasi nantinya mulai mendengarkan nama peristiwa tersebut, instans tersebut akan memeriksa buffer untuk pesan peristiwa dan memicu tugas yang menunggunya.

Catatan

Jika tidak ada instans orkestrasi dengan ID instans yang ditentukan, pesan kejadian akan dibuang.

HTTP

Berikut adalah contoh permintaan HTTP yang menimbulkan kejadian "Persetujuan" ke instans orkestrasi.

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

"true"

Dalam hal ini, ID instans di-hardcoded sebagai MyInstanceId.

Langkah berikutnya

Pelajari cara menerapkan penanganan kesalahan

Menjalankan sampel yang menunggu interaksi manusia