Привязки для устойчивых функций в Функциях Azure

Устойчивые функции поддерживают обе версии модели программирования Python для функций Azure. Так как Python v2 является рекомендуемой, примеры в этой статье представлены исключительно в этой версии.

Необходимые условия

• пакет SDK Устойчивые функции, который является пакетом Индекса пакетов Python (PyPI) azure-functions-durable, версии 1.2.2 или более поздней
• Пакет расширений версии 4.x (или более поздняя версия), который устанавливается в файле проекта host.json

Вы можете предоставить отзывы и предложения в пакете SDK Устойчивые функции для репозитория Python.

При разработке функций в .NET используйте атрибут OrchestrationTriggerAttribute .NET для настройки триггера оркестрации.

Для Java вы используете аннотацию @DurableOrchestrationTrigger для настройки триггера оркестрации.

При использовании версии 4 модели программирования Node.js для разработки функций вы импортируете app объект из @azure/functions npm модуля. Затем вы вызываете app.orchestration метод API устойчивых функций непосредственно в коде функции. Этот метод регистрирует функцию оркестратора с помощью платформы Устойчивые функции.

При записи функций оркестратора необходимо определить триггер оркестрации с помощью следующего объекта JSON в массиве bindings файла function.json :

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "orchestration": "<optional-name-of-orchestration>",
    "type": "orchestrationTrigger",
    "direction": "in"
}

Значением orchestration является имя оркестрации, которую клиенты должны использовать при запуске новых экземпляров функции оркестратора. Это свойство является необязательным. Если он не указан, используется имя функции.

При использовании модели программирования Python версии 2 можно определить триггер оркестрации с помощью декоратора orchestration_trigger непосредственно в коде функции Python.

В модели версии 2 вы получаете доступ к триггерам и привязкам в рамках Устойчивые функции из экземпляра DFApp. Этот подкласс FunctionApp можно использовать для экспорта декораторов, специфических для Устойчивые функции.

Атрибут, используемый для определения триггера, зависит от того, выполняются ли функции C#:

• В том же процессе, что и хост функций, или
• В изолированном рабочем процессе.

[FunctionName("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

[Function("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, string name)
{
    return $"Hello {name}!";
}

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    return `Hello ${name}`;
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.orchestration_trigger(context_name="context")
def my_orchestrator(context):
    result = yield context.call_activity("Hello", "Tokyo")
    return result

param($Context)

$InputData = $Context.Input
$InputData

@FunctionName("HelloWorldOrchestration")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    return String.format("Hello %s!", ctx.getInput(String.class));
}

Большинство функций оркестратора вызывают функции действия. Следующий код содержит пример Hello World, демонстрирующий вызов функции действия:

const { app } = require('@azure/functions');
const df = require('durable-functions');

const activityName = 'hello';

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    const result = yield context.df.callActivity(activityName, name);
    return result;
});

@FunctionName("HelloWorld")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String input = ctx.getInput(String.class);
    String result = ctx.callActivity("SayHello", input, String.class).await();
    return result;
}

Для настройки триггера действия используется атрибут ActivityTriggerAttribute .NET.

С помощью заметки @DurableActivityTrigger можно настроить триггер действия.

Чтобы зарегистрировать функцию действия, импортируйте app объект из @azure/functions npm модуля. Затем вы вызываете app.activity метод API устойчивых функций непосредственно в коде функции.

Чтобы определить триггер действия, используйте следующий объект JSON в массиве bindingsfunction.json:

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "activity": "<optional-name-of-activity>",
    "type": "activityTrigger",
    "direction": "in"
}

Значением activity является имя действия. Это значение — имя, используемое функциями оркестратора для вызова этой функции действия. Это свойство является необязательным. Если он не указан, используется имя функции.

Триггер действия можно определить с помощью декоратора activity_trigger непосредственно в коде функции Python.

const { app } = require('@azure/functions');
const df = require('durable-functions');
const activityName = 'hello';
df.app.activity(activityName, {
    handler: (input) => {
        return `Hello, ${input}`;
    },
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.activity_trigger(input_name="myInput")
def my_activity(myInput: str):
    return "Hello " + myInput

param($name)

"Hello $name!"

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

Например, функция действия может получать входные данные из функции оркестратора. Затем функция действия может отправлять эти входные данные в виде сообщения в Центры событий Azure.

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const input = context.df.getInput();
    yield context.df.callActivity('sendToEventHub', input);
    return `Message sent: ${input}`;
});

const { EventHubProducerClient } = require("@azure/event-hubs");
const connectionString = process.env.EVENT_HUB_CONNECTION_STRING;
const eventHubName = process.env.EVENT_HUB_NAME;

df.app.activity("sendToEventHub", {
    handler: async (message, context) => {
        const producer = new EventHubProducerClient(connectionString, eventHubName);
        try {
            const batch = await producer.createBatch();
            batch.tryAdd({ body: message });
            await producer.sendBatch(batch);
            context.log(`Message sent to Event Hubs: ${message}`);
        } catch (err) {
            context.log.error("Failed to send message to Event Hubs:", err);
            throw err;
        } finally {
            await producer.close();
        }
    },
});

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}'`);
    },
});

Вы можете привязаться к клиенту оркестрации с помощью атрибута DurableClientAttribute (OrchestrationClientAttribute в Устойчивые функции 1.x).

Вы можете привязаться к клиенту оркестрации с помощью аннотации @DurableClientInput.

Чтобы зарегистрировать клиентную функцию, импортируйте app объект из @azure/functions npm модуля. Затем вы вызываете метод API устойчивых функций, который зависит от типа триггера. Например, для триггера HTTP вызывается app.http метод. Для триггера очереди вызывается app.storageQueue метод.

Чтобы определить устойчивый триггер клиента, используйте следующий объект JSON в массиве bindingsfunction.json:

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "taskHub": "<optional-name-of-task-hub>",
    "connectionName": "<optional-name-of-connection-string-app-setting>",
    "type": "orchestrationClient",
    "direction": "in"
}

• Это taskHub свойство используется, если несколько приложений-функций используют одну и ту же учетную запись хранения, но их необходимо изолировать друг от друга. Если это свойство не указано, используется значение по умолчанию из host.json . Это значение должно соответствовать значению, используемому целевыми функциями оркестратора.
• Значение connectionName — это имя параметра приложения, содержащего строку подключения учетной записи хранения. Учетная запись хранения, представленная этой строкой подключения, должна совпадать с той, которую используют целевые функции оркестратора. Если это свойство не указано, используется строка подключения учетной записи хранения по умолчанию для функционального приложения.

Вы можете определить устойчивый триггер клиента с помощью декоратора durable_client_input непосредственно в коде функции Python.

Обычно выполняется привязка к реализации IDurableClient (DurableOrchestrationClient в Устойчивые функции версии 1.x), которая обеспечивает полный доступ ко всем API клиента оркестрации, поддерживаемым Устойчивые функции.

Обычно вы привязываетесь к классу DurableClientContext.

Чтобы получить доступ к объекту клиента, необходимо использовать пакет SDK для конкретного языка.

const { app } = require('@azure/functions');
const df = require('durable-functions');

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}' from queue message.`);
    },
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.queue_trigger(
    arg_name="msg",
    queue_name="start-orchestration",
    connection="AzureWebJobsStorage"
)
@myApp.durable_client_input(client_name="client")
async def client_function(msg: func.QueueMessage, client: df.DurableOrchestrationClient):
    input_data = msg.get_body().decode("utf-8")
    await client.start_new("my_orchestrator", None, input_data)
    return None

function.json

{
  "bindings": [
    {
      "name": "InputData",
      "type": "queueTrigger",
      "queueName": "durable-function-trigger",
      "direction": "in"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

run.ps1

param([string]$InputData, $TriggerMetadata)

$InstanceId = Start-DurableOrchestration -FunctionName 'HelloWorld' -Input $InputData

@FunctionName("QueueStart")
public void queueStart(
        @QueueTrigger(name = "input", queueName = "durable-function-trigger", connection = "Storage") String input,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    // Orchestration input comes from the queue message content.
    durableContext.getClient().scheduleNewOrchestrationInstance("HelloWorld", input);
}

Для настройки триггера сущности используется атрибут EntityTriggerAttribute .NET.

Чтобы зарегистрировать триггер сущности, импортируйте объект app из модуля @azure/functions npm. Затем вы вызываете app.entity метод API устойчивых функций непосредственно в коде функции.

const df = require('durable-functions');
df.app.entity('counter', (context) => {
    const currentValue = context.df.getState(() => 0);
    switch (context.df.operationName) {
        case 'add':
            context.df.setState(currentValue + context.df.getInput());
            break;
        case 'reset':
            context.df.setState(0);
            break;
        case 'get':
            context.df.return(currentValue);
            break;
    }
});

Триггер сущности можно определить с помощью заметки @DurableEntityTrigger . Используйте EntityRunner.loadAndRun, чтобы направлять операции в класс вашей сущности.

import com.microsoft.durabletask.AbstractTaskEntity;
import com.microsoft.durabletask.TaskEntityOperation;
import com.microsoft.durabletask.azurefunctions.DurableEntityTrigger;
import com.microsoft.durabletask.EntityRunner;

public class CounterEntity extends AbstractTaskEntity<Integer> {

    public void add(int value) {
        this.state += value;
    }

    public int get() {
        return this.state;
    }

    public void reset() {
        this.state = 0;
    }

    @Override
    protected Integer initializeState(TaskEntityOperation operation) {
        return 0;
    }

    @Override
    protected Class<Integer> getStateType() {
        return Integer.class;
    }
}

// Entity function
@FunctionName("Counter")
public String counterEntity(
    @DurableEntityTrigger(name = "req") String req) {
    return EntityRunner.loadAndRun(req, CounterEntity::new);
}

Триггер сущности можно определить с помощью декоратора entity_trigger непосредственно в коде функции Python.

Вы можете привязаться к клиенту сущности с помощью атрибута DurableClientAttribute .NET в функциях библиотеки классов .NET.

Вместо регистрации клиента сущности, вы используете signalEntity или callEntity, чтобы вызвать метод триггера сущности из любой зарегистрированной функции.

• Из функции, триггеруемой очередью, можно использовать client.signalEntity:

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  app.storageQueue('helloQueueStart', {
      queueName: 'start-orchestration',
      extraInputs: [df.input.durableClient()],
      handler: async (message, context) => {
          const client = df.getClient(context);
          const entityId = new df.EntityId('counter', 'myCounter');
          await client.signalEntity(entityId, 'add', 5);
      },
  });
  

• Из функции оркестратора можно использовать context.df.callEntity:

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  df.app.orchestration('entityCaller', function* (context) {
      const entityId = new df.EntityId('counter', 'myCounter');
      yield context.df.callEntity(entityId, 'add', 5);
      yield context.df.callEntity(entityId, 'add', 5);
      const result = yield context.df.callEntity(entityId, 'get');
      return result;
  });
  

Клиентский объект сущности можно определить с помощью декоратора durable_client_input непосредственно в коде функции Python.

Можно использовать аннотацию @DurableClientInput для привязки к DurableClientContext, который предоставляет методы для отправки сигналов сущностям и чтения состояния сущностей.

• Отправка сигнала сущности из функции, запускаемой по HTTP-триггеру:

  @FunctionName("SignalCounter")
  public HttpResponseMessage signalCounter(
      @HttpTrigger(name = "req", methods = {HttpMethod.POST},
          authLevel = AuthorizationLevel.ANONYMOUS)
      HttpRequestMessage<Void> request,
      @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
  
      EntityInstanceId entityId = new EntityInstanceId("Counter", "myCounter");
      durableContext.signalEntity(entityId, "add", 5);
  
      return request.createResponseBuilder(HttpStatus.ACCEPTED)
          .body("Signal sent")
          .build();
  }
  

• Чтение состояния сущности из функции, запускаемой по HTTP-запросу:

  @FunctionName("GetCounter")
  public HttpResponseMessage getCounter(
      @HttpTrigger(name = "req", methods = {HttpMethod.GET},
          authLevel = AuthorizationLevel.ANONYMOUS)
      HttpRequestMessage<Void> request,
      @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
  
      EntityInstanceId entityId = new EntityInstanceId("Counter", "myCounter");
      EntityMetadata metadata = durableContext.getEntityMetadata(entityId, true);
  
      if (metadata == null) {
          return request.createResponseBuilder(HttpStatus.NOT_FOUND).build();
      }
  
      Integer state = metadata.readStateAs(Integer.class);
      return request.createResponseBuilder(HttpStatus.OK)
          .body(state)
          .build();
  }