Anpassad orkestreringsstatus i Durable Functions (Azure Functions)

2023-06-01

Med anpassad orkestreringsstatus kan du ange ett anpassat statusvärde för orkestreringsfunktionen. Den här statusen tillhandahålls via HTTP GetStatus-API :et eller motsvarande SDK-API på orkestreringsklientobjektet.

Exempel på användningsfall

Visualisera förloppet

Klienter kan avsöka statusslutpunkten och visa ett förloppsgränssnitt som visualiserar det aktuella körningssteget. Följande exempel visar förloppsdelning:

Anteckning

Dessa C#-exempel skrivs för Durable Functions 2.x och är inte kompatibla med Durable Functions 1.x. Mer information om skillnaderna mellan olika versioner finns i artikeln Durable Functions versioner.

[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 orchestrator-funktion:

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 aktivitetsfunktion:

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

`E1_HelloSequence` Orchestrator-funktion

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` Aktivitetsfunktion

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

`E1_HelloSequence` Orchestrator-funktion

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` Aktivitetsfunktion

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

Och sedan får klienten endast utdata från orkestreringen när CustomStatus fältet är inställt på "London":

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

Anteckning

I JavaScript anges fältet customStatus när nästa yield eller return åtgärd schemaläggs.

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')

Anteckning

I Python anges fältet custom_status när nästa yield åtgärd eller return åtgärd schemaläggs.

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

Anpassning av utdata

Ett annat intressant scenario är segmentering av användare genom att returnera anpassade utdata baserat på unika egenskaper eller interaktioner. Med hjälp av anpassad orkestreringsstatus förblir koden på klientsidan allmän. Alla huvudändringar sker på serversidan enligt följande exempel:

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

Instruktionsspecifikation

Orkestreraren kan ge unika instruktioner till klienterna via det anpassade tillståndet. De anpassade statusinstruktionerna mappas till stegen i orkestreringskoden:

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

Fråga efter anpassad status med HTTP

I följande exempel visas hur anpassade statusvärden kan efterfrågas med hjälp av de inbyggda HTTP-API:erna.

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

När orkestreringen körs kan externa klienter hämta den här anpassade statusen:

GET /runtime/webhooks/durabletask/instances/instance123

Klienterna får följande svar:

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

Varning

Nyttolasten för anpassad status är begränsad till 16 KB UTF-16 JSON-text. Vi rekommenderar att du använder extern lagring om du behöver en större nyttolast.

Nästa steg

Lär dig mer om hållbara timers

Feedback

Var den här sidan till hjälp?

Dela via

Anpassad orkestreringsstatus i Durable Functions (Azure Functions)

Exempel på användningsfall

Visualisera förloppet

Anpassning av utdata

Instruktionsspecifikation

Fråga efter anpassad status med HTTP

Nästa steg

Feedback

Ytterligare resurser