Niestandardowy stan orkiestracji w Durable Functions (Azure Functions)

Niestandardowy stan orkiestracji umożliwia ustawienie niestandardowej wartości stanu dla funkcji orkiestratora. Ten stan jest udostępniany za pośrednictwem interfejsu API GetStatus PROTOKOŁU HTTP lub równoważnego interfejsu API zestawu SDK w obiekcie klienta aranżacji.

Przykładowe przypadki użycia

Wizualizowanie postępu

Klienci mogą sondować punkt końcowy stanu i wyświetlać interfejs użytkownika postępu, który wizualizuje bieżący etap wykonywania. W poniższym przykładzie pokazano udostępnianie postępu:

Uwaga

Te przykłady w języku C# są napisane dla Durable Functions 2.x i nie są zgodne z Durable Functions 1.x. Aby uzyskać więcej informacji na temat różnic między wersjami, zobacz artykuł Durable Functions versions (Wersje Durable Functions).

[FunctionName("E1_HelloSequence")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Tokyo"));
    context.SetCustomStatus("Tokyo");
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Seattle"));
    context.SetCustomStatus("Seattle");
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "London"));
    context.SetCustomStatus("London");

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

[FunctionName("E1_SayHello")]
public static string SayHello([ActivityTrigger] string name)
{
    return $"Hello {name}!";
}

E1_HelloSequence funkcja orkiestratora:

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

module.exports = df.orchestrator(function*(context){
    const outputs = [];

    outputs.push(yield context.df.callActivity("E1_SayHello", "Tokyo"));
    context.df.setCustomStatus("Tokyo");
    outputs.push(yield context.df.callActivity("E1_SayHello", "Seattle"));
    context.df.setCustomStatus("Seattle");
    outputs.push(yield context.df.callActivity("E1_SayHello", "London"));
    context.df.setCustomStatus("London");

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
});

E1_SayHello funkcja działania:

module.exports = async function(context, name) {
    return `Hello ${name}!`;
};

`E1_HelloSequence` Orchestrator, funkcja

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):
    
    output1 = yield context.call_activity('E1_SayHello', 'Tokyo')
    context.set_custom_status('Tokyo')
    output2 = yield context.call_activity('E1_SayHello', 'Seattle')
    context.set_custom_status('Seattle')
    output3 = yield context.call_activity('E1_SayHello', 'London')
    context.set_custom_status('London')
    
    return [output1, output2, output3]

main = df.Orchestrator.create(orchestrator_function)

`E1_SayHello` Funkcja działania

def main(name: str) -> str:
    return f"Hello {name}!"

`E1_HelloSequence` Orchestrator, funkcja

param($Context)

$output = @()

$output += Invoke-DurableActivity -FunctionName 'E1_SayHello' -Input 'Tokyo'
Set-DurableCustomStatus -CustomStatus 'Tokyo'

$output += Invoke-DurableActivity -FunctionName 'E1_SayHello' -Input 'Seattle'
Set-DurableCustomStatus -CustomStatus 'Seattle'

$output += Invoke-DurableActivity -FunctionName 'E1_SayHello' -Input 'London'
Set-DurableCustomStatus -CustomStatus 'London'


return $output

`E1_SayHello` Funkcja działania

param($name)

"Hello $name"

@FunctionName("HelloCities")
public String helloCitiesOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String result = "";
    result += ctx.callActivity("SayHello", "Tokyo", String.class).await() + ", ";
    ctx.setCustomStatus("Tokyo");
    result += ctx.callActivity("SayHello", "London", String.class).await() + ", ";
    ctx.setCustomStatus("London");
    result += ctx.callActivity("SayHello", "Seattle", String.class).await();
    ctx.setCustomStatus("Seattle");
    return result;
}

@FunctionName("SayHello")
public String sayHello(@DurableActivityTrigger(name = "name") String name) {
    return String.format("Hello %s!", name);
}

Następnie klient otrzyma dane wyjściowe orkiestracji tylko wtedy, gdy CustomStatus pole jest ustawione na "Londyn":

[FunctionName("HttpStart")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient starter,
    string functionName,
    ILogger log)
{
    // Function input comes from the request content.
    dynamic eventData = await req.Content.ReadAsAsync<object>();
    string instanceId = await starter.StartNewAsync(functionName, (string)eventData);

    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

    DurableOrchestrationStatus durableOrchestrationStatus = await starter.GetStatusAsync(instanceId);
    while (durableOrchestrationStatus.CustomStatus.ToString() != "London")
    {
        await Task.Delay(200);
        durableOrchestrationStatus = await starter.GetStatusAsync(instanceId);
    }

    HttpResponseMessage httpResponseMessage = new HttpResponseMessage(HttpStatusCode.OK)
    {
        Content = new StringContent(JsonConvert.SerializeObject(durableOrchestrationStatus))
    };

    return httpResponseMessage;
  }
}

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

module.exports = async function(context, req) {
    const client = df.getClient(context);

    // Function input comes from the request content.
    const eventData = req.body;
    const instanceId = await client.startNew(req.params.functionName, undefined, eventData);

    context.log(`Started orchestration with ID = '${instanceId}'.`);

    let durableOrchestrationStatus = await client.getStatus(instanceId);
    while (durableOrchestrationStatus.customStatus.toString() !== "London") {
        await new Promise((resolve) => setTimeout(resolve, 200));
        durableOrchestrationStatus = await client.getStatus(instanceId);
    }

    const httpResponseMessage = {
        status: 200,
        body: JSON.stringify(durableOrchestrationStatus),
    };

    return httpResponseMessage;
};

Uwaga

W języku JavaScript customStatus pole zostanie ustawione po zaplanowaniu następnej yield akcji lub return .

import json
import logging
import azure.functions as func
import azure.durable_functions as df
from time import sleep

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)    
    instance_id = await client.start_new(req.params.functionName, None, None)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")

    durable_orchestration_status = await client.get_status(instance_id)
    while durable_orchestration_status.custom_status != 'London':
        sleep(0.2)
        durable_orchestration_status = await client.get_status(instance_id)

    return func.HttpResponse(body='Success', status_code=200, mimetype='application/json')

Uwaga

W języku Python custom_status pole zostanie ustawione po zaplanowaniu następnego yield lub return akcji.

@FunctionName("StartHelloCities")
public HttpResponseMessage startHelloCities(
        @HttpTrigger(name = "req") HttpRequestMessage<Void> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        final ExecutionContext context) throws InterruptedException {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance("HelloCities");
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);

    OrchestrationMetadata metadata;
    try {
        metadata = client.waitForInstanceStart(instanceId, Duration.ofMinutes(5), true);
    } catch (TimeoutException ex) {
        return req.createResponseBuilder(HttpStatus.INTERNAL_SERVER_ERROR).build();
    }

    while (metadata.readCustomStatusAs(String.class) != "London") {
        Thread.sleep(200);
        metadata = client.getInstanceMetadata(instanceId, true);
    }

    return req.createResponseBuilder(HttpStatus.OK).build();
}

Dostosowywanie danych wyjściowych

Innym interesującym scenariuszem jest segmentowanie użytkowników przez zwracanie dostosowanych danych wyjściowych na podstawie unikatowych cech lub interakcji. Dzięki użyciu niestandardowego stanu orkiestracji kod po stronie klienta pozostanie ogólny. Wszystkie główne modyfikacje zostaną wprowadzone po stronie serwera, jak pokazano w poniższym przykładzie:

[FunctionName("CityRecommender")]
public static void Run(
  [OrchestrationTrigger] IDurableOrchestrationContext context)
{
  int userChoice = context.GetInput<int>();

  switch (userChoice)
  {
    case 1:
    context.SetCustomStatus(new
    {
      recommendedCities = new[] {"Tokyo", "Seattle"},
      recommendedSeasons = new[] {"Spring", "Summer"}
     });
      break;
    case 2:
      context.SetCustomStatus(new
      {
        recommendedCities = new[] {"Seattle, London"},
        recommendedSeasons = new[] {"Summer"}
      });
        break;
      case 3:
      context.SetCustomStatus(new
      {
        recommendedCities = new[] {"Tokyo, London"},
        recommendedSeasons = new[] {"Spring", "Summer"}
      });
        break;
  }

  // Wait for user selection and refine the recommendation
}

`CityRecommender` Orchestrator

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

module.exports = df.orchestrator(function*(context) {
    const userChoice = context.df.getInput();

    switch (userChoice) {
        case 1:
            context.df.setCustomStatus({
                recommendedCities: [ "Tokyo", "Seattle" ],
                recommendedSeasons: [ "Spring", "Summer" ],
            });
            break;
        case 2:
            context.df.setCustomStatus({
                recommendedCities: [ "Seattle", "London" ],
                recommendedSeasons: [ "Summer" ],
            });
            break;
        case 3:
            context.df.setCustomStatus({
                recommendedCity: [ "Tokyo", "London" ],
                recommendedSeasons: [ "Spring", "Summer" ],
            });
            break;
    }

    // Wait for user selection and refine the recommendation
});

`CityRecommender` Orchestrator

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    userChoice = int(context.get_input())

    if userChoice == 1:
        context.set_custom_status({
            'recommendedCities' : ['Tokyo', 'Seattle'], 
            'recommendedSeasons' : ['Spring', 'Summer']
        }))
    if userChoice == 2:
        context.set_custom_status({
            'recommendedCities' : ['Seattle', 'London']
            'recommendedSeasons' : ['Summer']
        }))
    if userChoice == 3:
        context.set_custom_status({
            'recommendedCities' : ['Tokyo', 'London'], 
            'recommendedSeasons' : ['Spring', 'Summer']
        }))



    # Wait for user selection and refine the recommendation

main = df.Orchestrator.create(orchestrator_function)

`CityRecommender` Orchestrator

param($Context)

$userChoice = $Context.Input -as [int]

if ($userChoice -eq 1) {
    Set-DurableCustomStatus -CustomStatus @{ recommendedCities = @('Tokyo', 'Seattle'); 
                                             recommendedSeasons = @('Spring', 'Summer') 
                                            }  
}

if ($userChoice -eq 2) {
    Set-DurableCustomStatus -CustomStatus @{ recommendedCities = @('Seattle', 'London'); 
                                             recommendedSeasons = @('Summer') 
                                            }  
}

if ($userChoice -eq 3) {
    Set-DurableCustomStatus -CustomStatus @{ recommendedCities = @('Tokyo', 'London'); 
                                             recommendedSeasons = @('Spring', 'Summer') 
                                            }  
}

# Wait for user selection and refine the recommendation

@FunctionName("CityRecommender")
public void cityRecommender(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    int userChoice = ctx.getInput(int.class);
    switch (userChoice) {
        case 1:
            ctx.setCustomStatus(new Recommendation(
                    new String[]{ "Tokyo", "Seattle" },
                    new String[]{ "Spring", "Summer" }));
            break;
        case 2:
            ctx.setCustomStatus(new Recommendation(
                    new String[]{ "Seattle", "London" },
                    new String[]{ "Summer" }));
            break;
        case 3:
            ctx.setCustomStatus(new Recommendation(
                    new String[]{ "Tokyo", "London" },
                    new String[]{ "Spring", "Summer" }));
            break;
    }

    // Wait for user selection with an external event
}

class Recommendation {
    public Recommendation() { }

    public Recommendation(String[] cities, String[] seasons) {
        this.recommendedCities = cities;
        this.recommendedSeasons = seasons;
    }

    public String[] recommendedCities;
    public String[] recommendedSeasons;
}

Specyfikacja instrukcji

Orkiestrator może dostarczyć klientom unikatowe instrukcje za pośrednictwem stanu niestandardowego. Instrukcje dotyczące stanu niestandardowego zostaną zamapowane na kroki w kodzie orkiestracji:

[FunctionName("ReserveTicket")]
public static async Task<bool> Run(
  [OrchestrationTrigger] IDurableOrchestrationContext context)
{
  string userId = context.GetInput<string>();

  int discount = await context.CallActivityAsync<int>("CalculateDiscount", userId);

  context.SetCustomStatus(new
  {
    discount = discount,
    discountTimeout = 60,
    bookingUrl = "https://www.myawesomebookingweb.com",
  });

  bool isBookingConfirmed = await context.WaitForExternalEvent<bool>("BookingConfirmed");

  context.SetCustomStatus(isBookingConfirmed
    ? new {message = "Thank you for confirming your booking."}
    : new {message = "The booking was not confirmed on time. Please try again."});

  return isBookingConfirmed;
}

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

module.exports = df.orchestrator(function*(context) {
    const userId = context.df.getInput();

    const discount = yield context.df.callActivity("CalculateDiscount", userId);

    context.df.setCustomStatus({
        discount,
        discountTimeout = 60,
        bookingUrl = "https://www.myawesomebookingweb.com",
    });

    const isBookingConfirmed = yield context.df.waitForExternalEvent("bookingConfirmed");

    context.df.setCustomStatus(isBookingConfirmed
        ? { message: "Thank you for confirming your booking." }
        : { message: "The booking was not confirmed on time. Please try again." }
    );

    return isBookingConfirmed;
});

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    userId = int(context.get_input())

    discount = yield context.call_activity('CalculateDiscount', userId)

    status = { 'discount' : discount,
        'discountTimeout' : 60,
        'bookingUrl' : "https://www.myawesomebookingweb.com",
    }
    context.set_custom_status(status)

    is_booking_confirmed = yield context.wait_for_external_event('BookingConfirmed')
    context.set_custom_status({'message': 'Thank you for confirming your booking.'} if is_booking_confirmed 
        else {'message': 'The booking was not confirmed on time. Please try again.'})
    return is_booking_confirmed

main = df.Orchestrator.create(orchestrator_function)

param($Context)

$userId = $Context.Input -as [int]

$discount = Invoke-DurableActivity -FunctionName 'CalculateDiscount' -Input $userId

$status = @{
            discount = $discount;
            discountTimeout = 60;
            bookingUrl = "https://www.myawesomebookingweb.com"
            }

Set-DurableCustomStatus -CustomStatus $status

$isBookingConfirmed = Invoke-DurableActivity -FunctionName 'BookingConfirmed'

if ($isBookingConfirmed) {
    Set-DurableCustomStatus -CustomStatus @{message = 'Thank you for confirming your booking.'}
} else {
    Set-DurableCustomStatus -CustomStatus @{message = 'The booking was not confirmed on time. Please try again.'}
}

return $isBookingConfirmed

@FunctionName("ReserveTicket")
public boolean reserveTicket(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String userID = ctx.getInput(String.class);
    int discount = ctx.callActivity("CalculateDiscount", userID, int.class).await();
    ctx.setCustomStatus(new DiscountInfo(discount, 60, "https://www.myawesomebookingweb.com"));

    boolean isConfirmed = ctx.waitForExternalEvent("BookingConfirmed", boolean.class).await();
    if (isConfirmed) {
        ctx.setCustomStatus("Thank you for confirming your booking.");
    } else {
        ctx.setCustomStatus("There was a problem confirming your booking. Please try again.");
    }

    return isConfirmed;
}

class DiscountInfo {
    public DiscountInfo() { }
    public DiscountInfo(int discount, int discountTimeout, String bookingUrl) {
        this.discount = discount;
        this.discountTimeout = discountTimeout;
        this.bookingUrl = bookingUrl;
    }
    public int discount;
    public int discountTimeout;
    public String bookingUrl;
}

Wykonywanie zapytań dotyczących stanu niestandardowego za pomocą protokołu HTTP

W poniższym przykładzie pokazano, jak można wykonywać zapytania dotyczące niestandardowych wartości stanu przy użyciu wbudowanych interfejsów API PROTOKOŁU HTTP.

public static async Task SetStatusTest([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    // ...do work...

    // update the status of the orchestration with some arbitrary data
    var customStatus = new { nextActions = new [] {"A", "B", "C"}, foo = 2, };
    context.SetCustomStatus(customStatus);

    // ...do more work...
}

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

module.exports = df.orchestrator(function*(context) {
    // ...do work...

    // update the status of the orchestration with some arbitrary data
    const customStatus = { nextActions: [ "A", "B", "C" ], foo: 2, };
    context.df.setCustomStatus(customStatus);

    // ...do more work...
});

import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    # ...do work...

    custom_status = {'nextActions': ['A','B','C'], 'foo':2}
    context.set_custom_status(custom_status)

    # ...do more work...

main = df.Orchestrator.create(orchestrator_function)

param($Context)

# ...do work...

Set-DurableCustomStatus -CustomStatus @{ nextActions = @('A', 'B', 'C'); 
                                         foo = 2 
                                        }  

# ...do more work...

@FunctionName("MyCustomStatusOrchestrator")
public void myCustomStatusOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    // ... do work ...

    // update the status of the orchestration with some arbitrary data
    CustomStatusPayload payload = new CustomStatusPayload();
    payload.nextActions = new String[] { "A", "B", "C" };
    payload.foo = 2;
    ctx.setCustomStatus(payload);

    // ... do more work ...
}

class CustomStatusPayload {
    public String[] nextActions;
    public int foo;
}

Gdy orkiestracja jest uruchomiona, klienci zewnętrzni mogą pobrać ten stan niestandardowy:

GET /runtime/webhooks/durabletask/instances/instance123

Klienci otrzymają następującą odpowiedź:

{
  "runtimeStatus": "Running",
  "input": null,
  "customStatus": { "nextActions": ["A", "B", "C"], "foo": 2 },
  "output": null,
  "createdTime": "2019-10-06T18:30:24Z",
  "lastUpdatedTime": "2019-10-06T19:40:30Z"
}

Ostrzeżenie

Niestandardowy ładunek stanu jest ograniczony do 16 KB tekstu JSON UTF-16. Zalecamy użycie magazynu zewnętrznego, jeśli potrzebujesz większego ładunku.

Następne kroki

Dowiedz się więcej o trwałych czasomierzach

Sprzężenie zwrotne

Czy ta strona była pomocna?

Last updated on 2023-06-01

Udostępnij przez

Niestandardowy stan orkiestracji w Durable Functions (Azure Functions)

Przykładowe przypadki użycia

Wizualizowanie postępu

Dostosowywanie danych wyjściowych

Specyfikacja instrukcji

Wykonywanie zapytań dotyczących stanu niestandardowego za pomocą protokołu HTTP

Następne kroki

Sprzężenie zwrotne

Dodatkowe źródła