Concatenamento delle funzioni

• Completare l'articolo introduttivo
• Clone o scaricare il progetto di esempi da GitHub

• Completare l'articolo introduttivo
• Clone o scaricare il progetto di esempi da GitHub

• Completare l'articolo introduttivo
• Clone o scaricare il progetto di esempi da GitHub

• Completare l'articolo introduttivo

[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"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello_DirectInput", "London"));

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

Tutte le funzioni di orchestrazione C# devono avere un parametro di tipo DurableOrchestrationContext, che esiste nell'assembly Microsoft.Azure.WebJobs.Extensions.DurableTask. Questo contesto di ambiente consente di chiamare altre funzioni dell’attività e di passare i parametri di input usando il metodo CallActivityAsync.

Il codice chiama E1_SayHello tre volte in sequenza con valori dei parametri diversi. Il valore restituito di ogni chiamata viene aggiunto all'elenco outputs, che viene restituito al termine della funzione.

function.json

Se si usa Visual Studio Code o il portale di Azure per lo sviluppo, ecco il contenuto del file function.json per la funzione dell'agente di orchestrazione. La maggior parte dei file function.json dell'agente di orchestrazione sono quasi identici a questo.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

Fondamentale è il tipo di associazione orchestrationTrigger. Tutte le funzioni di orchestrazione devono usare questo tipo di trigger.

__init__.py

La funzione dell'orchestratore è la seguente:

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


def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('E1_SayHello', "Tokyo")
    result2 = yield context.call_activity('E1_SayHello', "Seattle")
    result3 = yield context.call_activity('E1_SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

Tutte le funzioni di orchestrazione Python devono includere il pacchetto durable-functions. Si tratta di una libreria che consente di scrivere Durable Functions in Python. Due differenze principali tra una funzione dell'agente di orchestrazione e altre funzioni Python:

1. La funzione dell'agente di orchestrazione è una funzione del generatore.
2. Il file registra la funzione dell'agente di orchestrazione dichiarando main = df.Orchestrator.create(<orchestrator function name>) alla fine del file. Ciò consente di distinguerlo da altre funzioni helper dichiarate nel file.

L'oggetto context consente di chiamare altre funzioni dell’attività e di passare parametri di input tramite il metodo call_activity. Il codice chiama E1_SayHello tre volte in sequenza con valori di parametro diversi, usando yield per indicare che l'esecuzione deve attendere la restituzione delle chiamate di funzioni di attività asincrone. Il valore restituito di ogni chiamata viene restituito alla fine della funzione.

L'esempio di PowerShell non è ancora disponibile.

Java esempio non è ancora disponibile.

Questo codice mostra un agente di orchestrazione che chiama tre attività in sequenza e passa ogni output all'attività successiva:

using Microsoft.DurableTask;

[DurableTask]
public class GreetingOrchestration : TaskOrchestrator<string, string>
{
    public override async Task<string> RunAsync(TaskOrchestrationContext context, string name)
    {
        // Step 1: Say hello to the person
        string greeting = await context.CallActivityAsync<string>(nameof(SayHelloActivity), name);

        // Step 2: Process the greeting
        string processedGreeting = await context.CallActivityAsync<string>(nameof(ProcessGreetingActivity), greeting);

        // Step 3: Finalize the response
        string finalResponse = await context.CallActivityAsync<string>(nameof(FinalizeResponseActivity), processedGreeting);

        return finalResponse;
    }
}

Tutti gli orchestratori .NET ereditano da TaskOrchestrator<TInput, TOutput>. TaskOrchestrationContext Consente di chiamare le attività usando CallActivityAsync. Il codice chiama tre attività in sequenza, in cui ogni attività riceve l'output dell'attività precedente.

Il codice seguente mostra un agente di orchestrazione che chiama tre attività in sequenza:

import {
  OrchestrationContext,
  TOrchestrator,
} from "@microsoft/durabletask-js";

const greetingOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  name: string
): any {
  // Step 1: Say hello to the person
  const greeting: string = yield ctx.callActivity(sayHello, name);

  // Step 2: Process the greeting
  const processedGreeting: string = yield ctx.callActivity(processGreeting, greeting);

  // Step 3: Finalize the response
  const finalResponse: string = yield ctx.callActivity(finalizeResponse, processedGreeting);

  return finalResponse;
};

Tutti gli agenti di orchestrazione JavaScript sono funzioni generatori (async function*) che usano yield per chiamare le attività. Il metodo callActivity del contesto di orchestrazione pianifica l'esecuzione dell'attività. Il codice chiama tre attività in sequenza, passando l'output di ogni attività alla successiva.

Il codice seguente mostra un agente di orchestrazione che chiama tre attività in sequenza:

from durabletask import task

# Orchestrator function
def function_chaining_orchestrator(ctx: task.OrchestrationContext, name: str) -> str:
    """Orchestrator that demonstrates function chaining pattern."""

    # Call first activity
    greeting = yield ctx.call_activity(say_hello, input=name)

    # Call second activity with the result from first activity
    processed_greeting = yield ctx.call_activity(process_greeting, input=greeting)

    # Call third activity with the result from second activity
    final_response = yield ctx.call_activity(finalize_response, input=processed_greeting)

    return final_response

Tutti gli agenti di orchestrazione Python sono funzioni generatori che usano yield per chiamare le attività. Il metodo call_activity del contesto di orchestrazione pianifica l'esecuzione dell'attività. Il codice chiama tre attività in sequenza, passando l'output di ogni attività alla successiva.

Questo esempio viene illustrato per .NET, JavaScript, Java e Python.

Il codice seguente mostra un agente di orchestrazione che chiama tre attività in sequenza:

import com.microsoft.durabletask.*;

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "ActivityChaining"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                String input = ctx.getInput(String.class);

                // Call activities in sequence, passing output from one to the next
                String x = ctx.callActivity("Reverse", input, String.class).await();
                String y = ctx.callActivity("Capitalize", x, String.class).await();
                String z = ctx.callActivity("ReplaceWhitespace", y, String.class).await();

                ctx.complete(z);
            };
        }
    })
    .build();

In Java gli agenti di orchestrazione vengono definiti usando TaskOrchestrationFactory. Il metodo del contesto pianifica l'esecuzione dell'attività callActivity e await() attende il risultato. Il codice chiama tre attività in sequenza, passando l'output di ogni attività alla successiva.

[FunctionName("E1_SayHello")]
public static string SayHello([ActivityTrigger] IDurableActivityContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

Le attività usano l'attributo ActivityTrigger. Usare IDurableActivityContext per le azioni di attività, ad esempio la lettura dell'input con GetInput<T>.

E1_SayHello formatta una stringa di saluto.

Anziché eseguire l'associazione a IDurableActivityContext, eseguire l'associazione direttamente al tipo passato alla funzione dell'attività. Ad esempio:

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

E1_SayHello/function.json

Il file function.json per la funzione di attività E1_SayHello è simile a quello di E1_HelloSequence ad eccezione del fatto che usa un tipo di associazione activityTrigger anziché un tipo di associazione orchestrationTrigger.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "name",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

L'implementazione di E1_SayHello è un'operazione di formattazione delle stringhe relativamente semplice.

E1_SayHello/__init__.py

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

A differenza della funzione di orchestrazione, una funzione dell’attività non richiede alcuna configurazione speciale. L'input passato alla funzione di orchestrazione è accessibile direttamente come parametro per la funzione.

L'esempio di PowerShell sarà presto disponibile.

Java esempio presto disponibile.

Le attività in Durable Task SDK ereditano da TaskActivity<TInput, TOutput>:

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;

[DurableTask]
public class SayHelloActivity : TaskActivity<string, string>
{
    private readonly ILogger<SayHelloActivity> _logger;

    public SayHelloActivity(ILogger<SayHelloActivity> logger)
    {
        _logger = logger;
    }

    public override Task<string> RunAsync(TaskActivityContext context, string name)
    {
        _logger.LogInformation("Activity SayHello called with name: {Name}", name);
        return Task.FromResult($"Hello {name}!");
    }
}

[DurableTask]
public class ProcessGreetingActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string greeting)
    {
        return Task.FromResult($"{greeting} How are you today?");
    }
}

[DurableTask]
public class FinalizeResponseActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string response)
    {
        return Task.FromResult($"{response} I hope you're doing well!");
    }
}

Utilizzare l'inserimento delle dipendenze per ottenere servizi come ILogger. Aggiungere l'attributo [DurableTask] per registrare l'attività con il worker.

Le attività in Durable Task SDK sono funzioni semplici:

import { ActivityContext } from "@microsoft/durabletask-js";

const sayHello = async (_ctx: ActivityContext, name: string): Promise<string> => {
  return `Hello ${name}!`;
};

const processGreeting = async (_ctx: ActivityContext, greeting: string): Promise<string> => {
  return `${greeting} How are you today?`;
};

const finalizeResponse = async (_ctx: ActivityContext, response: string): Promise<string> => {
  return `${response} I hope you're doing well!`;
};

A differenza degli agenti di orchestrazione, le attività possono eseguire operazioni di I/O come chiamate HTTP, query di database e accesso ai file. L'input viene passato direttamente come parametro.

Le attività in Durable Task SDK sono funzioni semplici:

from durabletask import task

def say_hello(ctx: task.ActivityContext, name: str) -> str:
    """First activity that greets the user."""
    return f"Hello {name}!"

def process_greeting(ctx: task.ActivityContext, greeting: str) -> str:
    """Second activity that processes the greeting."""
    return f"{greeting} How are you today?"

def finalize_response(ctx: task.ActivityContext, response: str) -> str:
    """Third activity that finalizes the response."""
    return f"{response} I hope you're doing well!"

A differenza degli agenti di orchestrazione, le attività possono eseguire operazioni di I/O come chiamate HTTP, query di database e accesso ai file. L'input viene passato direttamente come parametro.

Questo esempio viene illustrato per .NET, JavaScript, Java e Python.

Le attività in Java vengono definite usando TaskActivityFactory:

import com.microsoft.durabletask.*;

.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Reverse"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            StringBuilder builder = new StringBuilder(input);
            builder.reverse();
            return builder.toString();
        };
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Capitalize"; }

    @Override
    public TaskActivity create() {
        return ctx -> ctx.getInput(String.class).toUpperCase();
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "ReplaceWhitespace"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            return input.trim().replaceAll("\\s", "-");
        };
    }
})

Registrare ogni attività con il generatore di ruoli di lavoro usando addActivity. Le attività possono eseguire operazioni di I/O e restituire risultati all'agente di orchestrazione.

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

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

        return starter.CreateCheckStatusResponse(req, instanceId);
    }
}

Per interagire con gli orchestratori, aggiungere un'associazione DurableClient di input. Usare il client per avviare un'orchestrazione e restituire una risposta HTTP che include GLI URL per controllare lo stato della nuova orchestrazione.

HttpStart/function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "route": "orchestrators/{functionName}",
      "methods": [
        "post",
        "get"
      ]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

Per interagire con gli orchestratori, la funzione deve includere un'associazione di input durableClient.

HttpStart/__init__.py

import logging

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


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

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

    return client.create_check_status_response(req, instance_id)

Usare il costruttore DurableOrchestrationClient per creare un client Durable Functions. Usare il client per avviare un'orchestrazione e restituire una risposta HTTP che include GLI URL per controllare lo stato della nuova orchestrazione.

L'esempio di PowerShell sarà presto disponibile.

Java esempio presto disponibile.

using Microsoft.DurableTask.Client;

// Create the client
var client = DurableTaskClientBuilder.UseDurableTaskScheduler(connectionString).Build();

// Schedule a new orchestration instance
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(GreetingOrchestration),
    input: "World");

Console.WriteLine($"Started orchestration with ID: {instanceId}");

// Wait for the orchestration to complete
OrchestrationMetadata result = await client.WaitForInstanceCompletionAsync(
    instanceId,
    getInputsAndOutputs: true);

Console.WriteLine($"Orchestration completed with result: {result.ReadOutputAs<string>()}");

Creare il DurableTaskClient usando una stringa di connessione al pianificatore di attività durevole. Usare ScheduleNewOrchestrationInstanceAsync per avviare un'orchestrazione e WaitForInstanceCompletionAsync attendere il completamento.

import {
  DurableTaskAzureManagedClientBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const connectionString =
  process.env.DURABLE_TASK_SCHEDULER_CONNECTION_STRING ||
  "Endpoint=http://localhost:8080;Authentication=None;TaskHub=default";

const client = new DurableTaskAzureManagedClientBuilder()
  .connectionString(connectionString)
  .build();

// Schedule a new orchestration instance
const instanceId = await client.scheduleNewOrchestration(greetingOrchestrator, "World");
console.log(`Started orchestration with ID: ${instanceId}`);

// Wait for the orchestration to complete
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);
console.log(`Orchestration completed with result: ${state?.serializedOutput}`);

Creare il DurableTaskAzureManagedClientBuilder usando una stringa di connessione al pianificatore di attività durevole. Usare scheduleNewOrchestration per avviare un'orchestrazione e waitForOrchestrationCompletion attendere il completamento.

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Create the client
client = DurableTaskSchedulerClient(
    host_address=endpoint,
    secure_channel=endpoint != "http://localhost:8080",
    taskhub=taskhub_name,
    token_credential=credential
)

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(
    function_chaining_orchestrator,
    input="World"
)

print(f"Started orchestration with ID: {instance_id}")

# Wait for the orchestration to complete
result = client.wait_for_orchestration_completion(instance_id, timeout=60)

if result and result.runtime_status == OrchestrationStatus.COMPLETED:
    print(f"Orchestration completed with result: {result.serialized_output}")

DurableTaskSchedulerClient si connette al pianificatore di attività durevoli. Usare schedule_new_orchestration per avviare un'orchestrazione e wait_for_orchestration_completion attendere il completamento.

Questo esempio viene illustrato per .NET, JavaScript, Java e Python.

import com.microsoft.durabletask.*;
import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientExtensions;

// Create the client
DurableTaskClient client = DurableTaskSchedulerClientExtensions
    .createClientBuilder(connectionString)
    .build();

// Schedule a new orchestration instance
String instanceId = client.scheduleNewOrchestrationInstance(
    "ActivityChaining",
    new NewOrchestrationInstanceOptions().setInput("Hello, world!"));

System.out.println("Started orchestration with ID: " + instanceId);

// Wait for the orchestration to complete
OrchestrationMetadata result = client.waitForInstanceCompletion(
    instanceId,
    Duration.ofSeconds(30),
    true);

System.out.println("Orchestration completed with result: " + result.readOutputAs(String.class));

Creare il DurableTaskClient usando un connection string. Usare scheduleNewOrchestrationInstance per avviare un'orchestrazione e waitForInstanceCompletion attendere il completamento.