Encadeamento de funções

• Conclua o artigo de início rápido
• Clone ou descarregue o projeto de exemplos do GitHub

• Conclua o artigo de início rápido
• Clone ou descarregue o projeto de exemplos do GitHub

• Conclua o artigo de início rápido
• Clone ou descarregue o projeto de exemplos do GitHub

• Conclua o artigo de início rápido

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

Todas as funções de orquestração em C# devem ter um parâmetro do tipo DurableOrchestrationContext, que existe no conjunto Microsoft.Azure.WebJobs.Extensions.DurableTask. Este objeto de contexto permite chamar outras funções de atividade e passar parâmetros de entrada usando seu CallActivityAsync método.

O código chama E1_SayHello três vezes em sequência com diferentes valores de parâmetro. O valor de retorno de cada chamada é adicionado à outputs lista, que é retornada no final da função.

function.json

Se usares Visual Studio Code ou o portal Azure para desenvolvimento, aqui está o conteúdo do ficheiro function.json para a função de orquestrador. A maioria dos ficheiros orquestrador function.json se parecem quase exatamente com isto.

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

O importante é o orchestrationTrigger tipo de ligação. Todas as funções do orquestrador devem usar esse tipo de gatilho.

__init__.py

Aqui está a função de orquestração:

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)

Todas as funções de orquestração Python devem incluir o pacote durable-functions. É uma biblioteca que permite escrever Durable Functions em Python. Duas diferenças-chave entre uma função de orquestrador e outras funções em Python:

1. A função orquestradora é uma função geradora.
2. O ficheiro regista a função do orquestrador indicando main = df.Orchestrator.create(<orchestrator function name>) no final do ficheiro. Isto ajuda a distingui-lo de outras funções auxiliares declaradas no ficheiro.

O context objeto permite chamar outras funções de atividade e passar parâmetros de entrada usando seu call_activity método. O código chama E1_SayHello três vezes em sequência com diferentes valores de parâmetro, usando yield para indicar que a execução deve aguardar que as chamadas de função de atividade assíncrona sejam retornadas. O valor de retorno de cada chamada é retornado no final da função.

O exemplo do PowerShell ainda não está disponível.

O exemplo de Java ainda não está disponível.

Este código mostra um orquestrador que chama três atividades em sequência e passa cada saída para a atividade seguinte:

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

Todos .NET orquestradores herdam de TaskOrchestrator<TInput, TOutput>. O TaskOrchestrationContext permite-lhe chamar atividades usando CallActivityAsync. O código chama três atividades em sequência, onde cada atividade recebe o resultado da atividade anterior.

O código seguinte mostra um orquestrador que chama três atividades em sequência:

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

Todos os orquestradores JavaScript são funções geradoras (async function*) que usam yield para chamar atividades. O método do contexto de orquestração callActivity agenda a execução da atividade. O código chama três atividades em sequência, passando a saída de cada atividade para a seguinte.

O código seguinte mostra um orquestrador que chama três atividades em sequência:

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

Todos os orquestradores Python são funções geradoras que usam yield para chamar atividades. O método do contexto de orquestração call_activity agenda a execução da atividade. O código chama três atividades em sequência, passando a saída de cada atividade para a seguinte.

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

O código seguinte mostra um orquestrador que chama três atividades em sequência:

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

Em Java, os orquestradores são definidos usando TaskOrchestrationFactory. O método do callActivity contexto agenda a execução da atividade e await() espera pelo resultado. O código chama três atividades em sequência, passando a saída de cada atividade para a seguinte.

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

As atividades usam o ActivityTrigger atributo. Use IDurableActivityContext para ações de atividade, como ler entrada com GetInput<T>.

E1_SayHello Formata uma cadeia de cumprimentos.

Em vez de ligar a IDurableActivityContext, liga diretamente ao tipo passado na função de atividade. Por exemplo:

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

E1_SayHello/function.json

O arquivo function.json para a função de atividade E1_SayHello é semelhante ao de E1_HelloSequence, exceto que ele usa um tipo de ligação activityTrigger em vez de um tipo de ligação orchestrationTrigger.

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

A implementação de E1_SayHello é uma operação de formatação de string relativamente trivial.

E1_SayHello/__init__.py

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

Ao contrário da função orquestradora, uma função de atividade não precisa de configuração especial. A entrada passada para a função pelo orquestrador é diretamente acessível como um parâmetro para a função.

Exemplo de PowerShell em breve.

Exemplo de Java em breve.

As atividades no SDK de Tarefas Duradouras herdam de 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!");
    }
}

Para obter serviços como ILogger, utilize a injeção de dependência. Adicione o [DurableTask] atributo para registar a atividade junto do trabalhador.

As atividades no SDK de Tarefas Duradouras são funções simples:

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

Ao contrário dos orquestradores, as atividades podem realizar operações de I/O como chamadas HTTP, consultas à base de dados e acesso a ficheiros. A entrada é passada diretamente como parâmetro.

As atividades no SDK de Tarefas Duradouras são funções simples:

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!"

Ao contrário dos orquestradores, as atividades podem realizar operações de I/O como chamadas HTTP, consultas à base de dados e acesso a ficheiros. A entrada é passada diretamente como parâmetro.

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

As atividades em Java são definidas 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", "-");
        };
    }
})

Registe cada atividade com o construtor de worker usando addActivity. As atividades podem realizar operações de I/O e devolver resultados ao orquestrador.

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

Para interagir com orquestradores, adicione uma DurableClient ligação de entrada. Use o cliente para iniciar uma orquestração e retorne uma resposta HTTP que inclua URLs para verificar o estado da nova orquestração.

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

Para interagir com orquestradores, a função deve incluir uma durableClient ligação de entrada.

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)

Use o construtor DurableOrchestrationClient para criar um cliente Durable Functions. Use o cliente para iniciar uma orquestração e retorne uma resposta HTTP que inclua URLs para verificar o estado da nova orquestração.

Exemplo de PowerShell em breve.

Exemplo de Java em breve.

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

Crie o DurableTaskClient usando uma cadeia de ligação com o Durable Task Scheduler. Use ScheduleNewOrchestrationInstanceAsync para iniciar uma orquestração e WaitForInstanceCompletionAsync esperar pela conclusão.

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

Crie o DurableTaskAzureManagedClientBuilder usando uma cadeia de ligação com o Durable Task Scheduler. Use scheduleNewOrchestration para iniciar uma orquestração e waitForOrchestrationCompletion esperar pela conclusão.

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

Liga-se DurableTaskSchedulerClient ao Durável Agendador de Tarefas. Use schedule_new_orchestration para iniciar uma orquestração e wait_for_orchestration_completion esperar pela conclusão.

Este exemplo é mostrado para .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));

Cria o DurableTaskClient usando um connection string. Use scheduleNewOrchestrationInstance para iniciar uma orquestração e waitForInstanceCompletion esperar pela conclusão.