永続エンティティ

エンティティ関数は、 永続エンティティと呼ばれる小さな状態を読み取って更新する操作を定義します。オーケストレーター関数と同様に、エンティティ関数はエンティティトリガーと呼ばれる特殊なトリガー型を使用 します。オーケストレーター関数とは異なり、エンティティ関数は、制御フローを介して状態を表すのではなく、エンティティの状態を明示的に管理します。エンティティは、それぞれが控えめな状態の多くのエンティティに作業を分散することで、アプリをスケールアウトするのに役立ちます。

エンティティは、永続エンティティと呼ばれる小さな状態を読み取って更新 する操作を定義します。オーケストレーターとは異なり、エンティティは制御フローを通じて状態を表すのではなく、状態を明示的に管理します。エンティティは、それぞれが控えめな状態の多くのエンティティに作業を分散することで、アプリをスケールアウトするのに役立ちます。

永続的エンティティのサポート

エンティティ関数および関連する機能は、Durable Functions 2.0 以降で使用できます。

プログラミング言語	永続的エンティティのサポート
隔離された.NET	✅
.NET インプロセス	✅
Java	❌
Python	✅
JavaScript	✅
PowerShell	❌

Durable Task SDK	永続的エンティティのサポート
隔離された.NET	✅
.NET インプロセス	✅
Java	✅
Python	✅
JavaScript	✅
PowerShell	❌

永続的エンティティの一般的な概念

エンティティは、メッセージを使用して通信する小さなサービスのように動作します。各エンティティには一意の ID と、必要に応じて内部状態があります。エンティティは、プロンプトが表示されたときに操作を実行します。操作は、状態の更新、外部サービスの呼び出し、または応答の待機を行う場合があります。エンティティは、ランタイムが信頼できるキューを介して送信するメッセージを介して、他のエンティティ、オーケストレーション、およびクライアントと通信します。

競合を防ぐために、1 つのエンティティが操作を順次実行します。

注

エンティティを呼び出すと、ペイロードが完了するように処理され、新しい入力が到着したときにアクティブ化される新しい実行がスケジュールされます。その結果、エンティティ実行ログには、各呼び出し後に追加の実行が表示される場合があります。それは期待されています。

エンティティ識別子

エンティティ ID を使用してエンティティにアクセスします。エンティティ ID は、エンティティインスタンスを一意に識別する文字列のペアです。構成は次のとおりです。

エンティティの種類を識別するエンティティ名。たとえば、「 Counter 」のように入力します。この名前は、エンティティを実装するエンティティ関数の名前と一致します。大文字と小文字は区別されません。
エンティティキー。同じ名前の他のすべてのエンティティ間でエンティティを一意に識別します。たとえば、GUID です。

たとえば、 Counter エンティティ関数は、オンラインゲームでスコアを維持するために使用できます。ゲームの各インスタンスには、 @Counter@Game1 や @Counter@Game2などの一意のエンティティ ID があります。エンティティを対象にするには、そのエンティティ ID を指定します。

エンティティの操作

エンティティに対して操作を呼び出すには、次のように指定します。

ターゲットエンティティのエンティティID。
操作名。実行する操作を指定する文字列です。たとえば、 Counter エンティティは、 add、 get、または reset 操作をサポートできます。
操作の入力。操作の省略可能なパラメーターです。たとえば、 add 操作は、整数の量を入力として受け取ります。
スケジュールされた時刻。これは、操作の配信時刻を指定する省略可能なパラメーターです。たとえば、数日後に実行するように操作をスケジュールします。

操作は、JavaScript エラーや.NET例外などの結果値またはエラー結果を返すことができます。呼び出し元のオーケストレーションは、結果またはエラーを受け取ります。

操作は、.NET例外やPython例外などの結果値またはエラー結果を返すことができます。呼び出し元は結果またはエラーを受け取ります。

エンティティ操作では、エンティティの状態の作成、読み取り、更新、および削除を行うこともできます。ランタイムは常にエンティティの状態をストレージに保持します。

エンティティの定義

.NETでエンティティを定義するには、次の 2 つの API のいずれかを使用します。

関数ベースの構文: 関数ベースの構文では、各エンティティを関数として記述し、アプリで操作をディスパッチします。この構文は、単純な状態のエンティティ、操作の数が少ないエンティティ、またはアプリフレームワークのような動的な一連の操作に適しています。ただし、コンパイル時に型エラーをキャッチしないため、保守が面倒な場合があります。

クラスベースの構文: クラスベースの構文では、.NETクラスとメソッドはエンティティと操作をモデル化します。この構文により、コードの読み取りが容易になり、型セーフな方法で操作を呼び出すことができます。これは関数ベースの構文の上に薄いレイヤーであるため、両方のバリアントを同じアプリに混在させることができます。

使用する API は、C# 関数の実行場所によって異なります。 分離ワーカープロセスをお勧めしますが、ホストプロセスで実行することもできます。

インプロセス関数ベースの例:

この例では、永続的な関数として実装される単純な Counter エンティティを示します。整数状態を使用する 3 つの操作 (add、 reset、 get) を定義します。

[FunctionName("Counter")]
public static void Counter([EntityTrigger] IDurableEntityContext ctx)
{
    switch (ctx.OperationName.ToLowerInvariant())
    {
        case "add":
            ctx.SetState(ctx.GetState<int>() + ctx.GetInput<int>());
            break;
        case "reset":
            ctx.SetState(0);
            break;
        case "get":
            ctx.Return(ctx.GetState<int>());
            break;
    }
}

詳細については、「関数ベースの構文」を参照してください。

インプロセスクラスベースの例:

この例では、クラスとメソッドを使用して実装された同じ Counter エンティティを示します。

[JsonObject(MemberSerialization.OptIn)]
public class Counter
{
    [JsonProperty("value")]
    public int CurrentValue { get; set; }

    public void Add(int amount) => this.CurrentValue += amount;

    public void Reset() => this.CurrentValue = 0;

    public int Get() => this.CurrentValue;

    [FunctionName(nameof(Counter))]
    public static Task Run([EntityTrigger] IDurableEntityContext ctx)
        => ctx.DispatchAsync<Counter>();
}

このエンティティは、現在のカウンター値を保持する Counter オブジェクトに状態を格納します。 Durable Functionsは、Json.NET ライブラリを使用して、このオブジェクトをシリアル化および逆シリアル化します。

詳細については、「エンティティクラスの定義」を参照してください。

分離されたワーカープロセス関数ベースの例:

次の例は、分離されたワーカープロセスの関数ベースの Counter エンティティを示しています。 add、reset、get、およびdeleteをサポートします。

[Function(nameof(Counter))]
public static Task Counter([EntityTrigger] TaskEntityDispatcher dispatcher)
{
    return dispatcher.DispatchAsync(operation =>
    {
        if (operation.State.GetState(typeof(int)) is null)
        {
            operation.State.SetState(0);
        }

        switch (operation.Name.ToLowerInvariant())
        {
            case "add":
                int state = operation.State.GetState<int>();
                state += operation.GetInput<int>();
                operation.State.SetState(state);
                return new(state);
            case "reset":
                operation.State.SetState(0);
                break;
            case "get":
                return new(operation.State.GetState<int>());
            case "delete":
                operation.State.SetState(null);
                break;
        }

        return default;
    });
}

分離されたワーカープロセスのクラスベースの例:

次の例は、クラスとメソッドを使用した Counter エンティティの実装を示しています。

public class Counter : TaskEntity<int>
{
    readonly ILogger logger;

    public Counter(ILogger<Counter> logger)
    {
        this.logger = logger;
    }

    public void Add(int amount) => this.State += amount;

    public void Reset() => this.State = 0;

    public int Get() => this.State;

    [Function(nameof(Counter))]
    public Task RunEntityAsync([EntityTrigger] TaskEntityDispatcher dispatcher)
    {
        return dispatcher.DispatchAsync(this);
    }
}

永続エンティティは、1.3.0 npm パッケージのバージョン durable-functions 以降の JavaScript で使用できます。この例では、JavaScript で永続的な関数として実装される Counter エンティティを示します。

カウンター/function.json

{
  "bindings": [
    {
      "name": "context",
      "type": "entityTrigger",
      "direction": "in"
    }
  ],
  "disabled": false
}

Counter/index.js

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

module.exports = df.entity(function(context) {
    const currentValue = context.df.getState(() => 0);
    switch (context.df.operationName) {
        case "add": {
            const amount = context.df.getInput();
            context.df.setState(currentValue + amount);
            break;
        }
        case "reset":
            context.df.setState(0);
            break;
        case "get":
            context.df.return(currentValue);
            break;
    }
});

次のコードは、Counter エンティティであり、Pythonで記述された永続的な関数として実装されています。

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

# Entity function called counter
@myApp.entity_trigger(context_name="context")
def Counter(context):
    current_value = context.get_state(lambda: 0)
    operation = context.operation_name
    if operation == "add":
        amount = context.get_input()
        current_value += amount
    elif operation == "reset":
        current_value = 0
    elif operation == "get":
        context.set_result(current_value)
    context.set_state(current_value)

Durable Task SDK for .NETでは、クラスベースの構文を使用したエンティティの定義がサポートされています。 TaskEntity<TState>基底クラスを実装して、エンティティを定義できます。

次の例は、Durable Task SDK を使用して実装された Counter エンティティを示しています。

using Microsoft.DurableTask.Entities;

public class Counter : TaskEntity<int>
{
    public void Add(int amount) => this.State += amount;

    public void Reset() => this.State = 0;

    public int Get() => this.State;
}

エンティティをワーカーに登録するには:

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddEntity<Counter>();
    })
    .UseDurableTaskScheduler(connectionString);

オーケストレーターからエンティティをシグナル通知または呼び出すには:

public class EntityOrchestration : TaskOrchestrator<string, int>
{
    public override async Task<int> RunAsync(TaskOrchestrationContext context, string entityKey)
    {
        var entityId = new EntityInstanceId(nameof(Counter), entityKey);

        // Signal the entity (fire-and-forget)
        await context.Entities.SignalEntityAsync(entityId, nameof(Counter.Add), 1);

        // Call the entity and wait for response
        int currentValue = await context.Entities.CallEntityAsync<int>(entityId, nameof(Counter.Get));

        return currentValue;
    }
}

Durable Task SDK for JavaScript/TypeScript では、クラスベースの構文を使用したエンティティの定義がサポートされています。 TaskEntity<TState>基底クラスを拡張して、エンティティを定義できます。

次の例は、Durable Task SDK を使用して実装された Counter エンティティを示しています。

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

class CounterEntity extends TaskEntity {
  add(amount) {
    this.state.value += amount;
    return this.state.value;
  }

  get() {
    return this.state.value;
  }

  reset() {
    this.state.value = 0;
  }

  initializeState() {
    return { value: 0 };
  }
}

エンティティをワーカーに登録するには:

worker.addNamedEntity("Counter", () => new CounterEntity());

durable Task SDK for Python では、関数ベースの構文を使用したエンティティの定義がサポートされています。

次の例は、Durable Task SDK を使用して実装された Counter エンティティを示しています。

from durabletask import task

def counter_entity(ctx: task.EntityContext):
    """Counter entity that maintains an integer state."""
    current_value = ctx.get_state(lambda: 0)
    operation = ctx.operation_name

    if operation == "add":
        amount = ctx.get_input()
        current_value += amount
    elif operation == "reset":
        current_value = 0
    elif operation == "get":
        ctx.set_result(current_value)

    ctx.set_state(current_value)

エンティティをワーカーに登録するには:

with DurableTaskSchedulerWorker(
    host_address=endpoint,
    secure_channel=secure_channel,
    taskhub=taskhub,
    token_credential=credential
) as worker:
    worker.add_entity(counter_entity)
    worker.start()

Durable Task SDK for Java では、クラスベースの構文を使用したエンティティの定義がサポートされています。 AbstractTaskEntity<TState>基底クラスを拡張して、エンティティを定義できます。

次の例は、Durable Task SDK を使用して実装された Counter エンティティを示しています。

import com.microsoft.durabletask.AbstractTaskEntity;
import com.microsoft.durabletask.TaskEntityOperation;

public class CounterEntity extends AbstractTaskEntity<Integer> {

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

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

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

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

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

エンティティをワーカーに登録するには:

DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder()
        .addEntity("Counter", CounterEntity::new)
        .build();

worker.start();

オーケストレーターからエンティティをシグナル通知または呼び出すには:

return ctx -> {
    EntityInstanceId counterId = new EntityInstanceId("Counter", "myCounter");

    // Signal the entity (fire-and-forget)
    ctx.signalEntity(counterId, "add", 5);

    // Call the entity and wait for response
    int currentValue = ctx.callEntity(counterId, "get", Integer.class).await();

    ctx.complete(currentValue);
};

エンティティをアクセスする

一方向または双方向の通信を使用してエンティティにアクセスする:

エンティティを呼び出すと、双方向 (ラウンドトリップ) 通信が使用されます。操作メッセージをエンティティに送信し、応答メッセージを待ってから続行します。応答メッセージは、結果値またはエラー (JavaScript エラーや.NET例外など) を提供します。
エンティティのシグナル通知は、一方向 (ファイアアンドフォーゲット、返信なしの自動処理) 通信を使用します。操作メッセージを送信しますが、応答を待機しないでください。ランタイムは配信を保証しますが、送信者は結果の値やエラーを観察できません。

クライアント関数、オーケストレーター関数、またはエンティティ関数からエンティティにアクセスします。すべてのコンテキストが両方の通信の種類をサポートしているわけではありません。

クライアント関数は、シグナル通知エンティティとエンティティの状態の読み取りをサポートします。
オーケストレーター関数は、シグナル通知と呼び出しエンティティをサポートします。
エンティティ関数は、シグナルエンティティをサポートします。

一方向または双方向の通信を使用してエンティティにアクセスする:

エンティティを呼び出すと、双方向 (ラウンドトリップ) 通信が使用されます。操作メッセージをエンティティに送信し、応答メッセージを待ってから続行します。応答メッセージは、結果の値またはエラーを提供します。
エンティティのシグナル通知は、一方向 (ファイアアンドフォーゲット、返信なしの自動処理) 通信を使用します。操作メッセージを送信しますが、応答を待機しないでください。ランタイムは配信を保証しますが、送信者は結果の値やエラーを観察できません。

クライアントまたはオーケストレーターからエンティティにアクセスします。すべてのコンテキストが両方の通信の種類をサポートしているわけではありません。

クライアントは、シグナル通知エンティティとエンティティの状態の読み取りをサポートします。
オーケストレーターは、シグナル通知と呼び出しエンティティをサポートします。

次の例は、エンティティにアクセスする方法を示しています。

例: クライアントがエンティティにシグナルを送信する

通常のAzure関数 (クライアント関数とも呼ばれます) からエンティティにアクセスするには、entity クライアントバインディングを使用します。次の例では、このバインドを使用してエンティティにシグナル通知する、キューによってトリガーされた関数を示します。

注

わかりやすくするために、次の例では、エンティティにアクセスするための緩やかに型指定された構文を示しています。一般に、より多くの型チェックが提供されるため、インターフェイスを介してエンティティにアクセスします。

処理中:

[FunctionName("AddFromQueue")]
public static Task Run(
    [QueueTrigger("durable-function-trigger")] string input,
    [DurableClient] IDurableEntityClient client)
{
    // Entity operation input comes from the queue message content.
    var entityId = new EntityId(nameof(Counter), "myCounter");
    int amount = int.Parse(input);
    return client.SignalEntityAsync(entityId, "Add", amount);
}

隔離されたワーカープロセス:

[Function("AddFromQueue")]
public static Task Run(
    [QueueTrigger("durable-function-trigger")] string input,
    [DurableClient] DurableTaskClient client)
{
    // Entity operation input comes from the queue message content.
    var entityId = new EntityInstanceId(nameof(Counter), "myCounter");
    int amount = int.Parse(input);
    return client.Entities.SignalEntityAsync(entityId, "Add", amount);
}

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

module.exports = async function (context) {
    const client = df.getClient(context);
    const entityId = new df.EntityId("Counter", "myCounter");
    await client.signalEntity(entityId, "add", 1);
};

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

# An HTTP-Triggered Function with a Durable Functions Client to set a value on a durable entity
@myApp.route(route="entitysetvalue")
@myApp.durable_client_input(client_name="client")
async def http_set(req: func.HttpRequest, client):
    entityId = df.EntityId("Counter", "myCounter")
    await client.signal_entity(entityId, "add", 1)
    return func.HttpResponse("Done", status_code=200)

シグナルという用語は、エンティティ API 呼び出しが一方向および非同期であることを意味します。クライアント関数は、エンティティが操作をいつ処理するかを認識できません。クライアント関数は、結果の値または例外を観察できません。

クライアントからエンティティにアクセスするには、 DurableTaskClient を使用してエンティティの状態を通知または読み取ります。

// Signal an entity
var entityId = new EntityInstanceId(nameof(Counter), "myCounter");
await client.Entities.SignalEntityAsync(entityId, nameof(Counter.Add), 1);

// Signal an entity
const counterId = new EntityInstanceId("Counter", "myCounter");
await client.signalEntity(counterId, "add", 1);

# Signal an entity
entity_id = EntityId("Counter", "myCounter")
client.signal_entity(entity_id, "add", 1)

// Signal an entity
EntityInstanceId counterId = new EntityInstanceId("Counter", "myCounter");
client.getEntities().signalEntity(counterId, "add", 1);

シグナルという用語は、エンティティ API 呼び出しが一方向および非同期であることを意味します。クライアントは、エンティティが操作をいつ処理するかを認識できません。

例: クライアントがエンティティの状態を読み取る

クライアント関数からエンティティの状態を照会します。

処理中:

[FunctionName("QueryCounter")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function)] HttpRequestMessage req,
    [DurableClient] IDurableEntityClient client)
{
    var entityId = new EntityId(nameof(Counter), "myCounter");
    EntityStateResponse<JObject> stateResponse = await client.ReadEntityStateAsync<JObject>(entityId);
    return req.CreateResponse(HttpStatusCode.OK, stateResponse.EntityState);
}

隔離されたワーカープロセス:

[Function("QueryCounter")]
public static async Task<HttpResponseData> Run(
    [HttpTrigger(AuthorizationLevel.Function)] HttpRequestData req,
    [DurableClient] DurableTaskClient client)
{
    var entityId = new EntityInstanceId(nameof(Counter), "myCounter");
    EntityMetadata<int>? entity = await client.Entities.GetEntityAsync<int>(entityId);

    if (entity is null)
    {
        return req.CreateResponse(HttpStatusCode.NotFound);
    }

    HttpResponseData response = req.CreateResponse(HttpStatusCode.OK);
    await response.WriteAsJsonAsync(entity);

    return response;
}

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

module.exports = async function (context) {
    const client = df.getClient(context);
    const entityId = new df.EntityId("Counter", "myCounter");
    const stateResponse = await client.readEntityState(entityId);
    return stateResponse.entityState;
};

# An HTTP-Triggered Function with a Durable Functions Client to retrieve the state of a durable entity
@myApp.route(route="entityreadvalue")
@myApp.durable_client_input(client_name="client")
async def http_read(req: func.HttpRequest, client):
    entityId = df.EntityId("Counter", "myCounter")
    entity_state_result = await client.read_entity_state(entityId)
    entity_state = "No state found"
    if entity_state_result.entity_exists:
      entity_state = str(entity_state_result.entity_state)
    return func.HttpResponse(entity_state)

クライアントはエンティティの状態を照会できます。

var entityId = new EntityInstanceId(nameof(Counter), "myCounter");
EntityMetadata<int>? entity = await client.Entities.GetEntityAsync<int>(entityId);

if (entity != null)
{
    Console.WriteLine($"Current value: {entity.State}");
}

const counterId = new EntityInstanceId("Counter", "myCounter");
const metadata = await client.getEntity(counterId);

if (metadata.exists) {
  console.log(`Current value: ${metadata.state?.value}`);
}

entity_id = EntityId("Counter", "myCounter")
state = client.get_entity_state(entity_id)
if state is not None:
    print(f"Current value: {state}")

EntityInstanceId counterId = new EntityInstanceId("Counter", "myCounter");
EntityMetadata entityMetadata = client.getEntities().getEntityMetadata(counterId, true);

if (entityMetadata != null) {
    Integer state = entityMetadata.readStateAs(Integer.class);
    System.out.printf("Current value: %d%n", state);
}

エンティティ状態クエリは永続トラッキングストアに送信され、エンティティの最新の保存状態が返されます。この状態は、常に "コミット済み" 状態です。つまり、操作の実行中に想定される一時的な中間状態ではありません。ただし、この状態は、エンティティのメモリ内状態と比較して古くなる可能性があります。次のセクションで説明するように、エンティティのメモリ内の状態を読み取ることができるのはオーケストレーションだけです。

例: オーケストレーションがエンティティをシグナル通知して呼び出す

オーケストレーター関数は、オーケストレーショントリガーバインドの API を使用してエンティティにアクセスできます。次のコード例は、 Counter エンティティを呼び出して通知するオーケストレーター関数を示しています。

処理中:

[FunctionName("CounterOrchestration")]
public static async Task Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var entityId = new EntityId(nameof(Counter), "myCounter");

    // Two-way call to the entity which returns a value - awaits the response
    int currentValue = await context.CallEntityAsync<int>(entityId, "Get");
    if (currentValue < 10)
    {
        // One-way signal to the entity which updates the value - does not await a response
        context.SignalEntity(entityId, "Add", 1);
    }
}

隔離されたワーカープロセス:

[Function("CounterOrchestration")]
public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext context)
{
    var entityId = new EntityInstanceId(nameof(Counter), "myCounter");

    // Two-way call to the entity which returns a value - awaits the response
    int currentValue = await context.Entities.CallEntityAsync<int>(entityId, "Get");

    if (currentValue < 10)
    {
        // One-way signal to the entity which updates the value - does not await a response
        await context.Entities.SignalEntityAsync(entityId, "Add", 1);
    }
}

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

module.exports = df.orchestrator(function* (context) {
    const entityId = new df.EntityId("Counter", "myCounter");

    // Two-way call to the entity which returns a value - awaits the response
    const currentValue = yield context.df.callEntity(entityId, "get");
    return currentValue;
});

注

JavaScript では、オーケストレーターからのエンティティのシグナル通知はサポートされていません。 callEntity を代わりに使用します。

@myApp.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext):
    entityId = df.EntityId("Counter", "myCounter")
    context.signal_entity(entityId, "add", 3)
    state = yield context.call_entity(entityId, "get")
    return state

エンティティを呼び出して応答を取得できるのはオーケストレーションだけです。これは戻り値または例外です。クライアントバインドを使用するクライアント関数は、エンティティのみを通知できます。

オーケストレーターは、コンテキストの Entities API を使用してエンティティにアクセスできます。

public class CounterOrchestration : TaskOrchestrator<string, int>
{
    public override async Task<int> RunAsync(TaskOrchestrationContext context, string entityKey)
    {
        var entityId = new EntityInstanceId(nameof(Counter), entityKey);

        // Two-way call to the entity which returns a value - awaits the response
        int currentValue = await context.Entities.CallEntityAsync<int>(entityId, nameof(Counter.Get));

        if (currentValue < 10)
        {
            // One-way signal to the entity - does not await a response
            await context.Entities.SignalEntityAsync(entityId, nameof(Counter.Add), 1);
        }

        return currentValue;
    }
}

const counterOrchestration = async function* (ctx, entityKey) {
  const counterId = new EntityInstanceId("Counter", entityKey);

  // Two-way call to the entity which returns a value - awaits the response
  const currentValue = yield* ctx.entities.callEntity(counterId, "get");

  if (currentValue < 10) {
    // One-way signal to the entity - does not await a response
    ctx.entities.signalEntity(counterId, "add", 1);
  }

  return currentValue;
};

def counter_orchestration(ctx: task.OrchestrationContext, entity_key: str):
    entity_id = EntityId("Counter", entity_key)

    # Two-way call to the entity which returns a value
    current_value = yield ctx.call_entity(entity_id, "get")

    if current_value < 10:
        # One-way signal to the entity
        ctx.signal_entity(entity_id, "add", 1)

    return current_value

return ctx -> {
    EntityInstanceId counterId = new EntityInstanceId("Counter", entityKey);

    // Two-way call to the entity which returns a value - awaits the response
    int currentValue = ctx.callEntity(counterId, "get", Integer.class).await();

    if (currentValue < 10) {
        // One-way signal to the entity - does not await a response
        ctx.signalEntity(counterId, "add", 1);
    }

    ctx.complete(currentValue);
};

エンティティを呼び出して応答を取得できるのはオーケストレーターだけです。これは戻り値または例外です。クライアントはエンティティのみを通知できます。

注

オーケストレーターからエンティティを呼び出すことは、アクティビティの呼び出しに似ています。主な違いは、エンティティはアドレス (エンティティ ID) を持つ永続オブジェクトであり、操作名の指定をサポートしていることです。アクティビティはステートレスであり、操作の概念を持っていません。

例: エンティティがエンティティにシグナルを送信する

エンティティ関数は、操作の実行中に他のエンティティ (またはそれ自体) にシグナルを送信することができます。たとえば、前の Counter エンティティの例を変更して、カウンターが 100 に達したときに "マイルストーンに達した" シグナルをモニターエンティティに送信します。

処理中:

   case "add":
        var currentValue = ctx.GetState<int>();
        var amount = ctx.GetInput<int>();
        if (currentValue < 100 && currentValue + amount >= 100)
        {
            ctx.SignalEntity(new EntityId("MonitorEntity", ""), "milestone-reached", ctx.EntityKey);
        }

        ctx.SetState(currentValue + amount);
        break;

隔離されたワーカープロセス:

case "add":
    var currentValue = operation.State.GetState<int>();
    var amount = operation.GetInput<int>();
    if (currentValue < 100 && currentValue + amount >= 100)
    {
        operation.Context.SignalEntity(new EntityInstanceId("MonitorEntity", ""), "milestone-reached", operation.Context.EntityInstanceId);
    }

    operation.State.SetState(currentValue + amount);
    break;

    case "add":
        const amount = context.df.getInput();
        if (currentValue < 100 && currentValue + amount >= 100) {
            const entityId = new df.EntityId("MonitorEntity", "");
            context.df.signalEntity(entityId, "milestone-reached", context.df.instanceId);
        }
        context.df.setState(currentValue + amount);
        break;

エンティティの調整

場合によっては、複数のエンティティ間で操作を調整する必要があります。たとえば、銀行業務アプリケーションでは、エンティティは個々の銀行口座を表すことができます。ある口座から別の口座に資金を送金する場合は、ソースアカウントに十分な資金があることを確認する必要があります。また、両方のアカウントを 1 つの一貫した操作として更新する必要もあります。

例: 資金の送金 (C#)

次のコード例では、オーケストレーター関数を使用して、2 つの口座エンティティ間で送金を行います。エンティティの更新を調整するには、 LockAsync メソッドを使用してオーケストレーションに 重要なセクション を作成します。

注

わかりやすくするために、この例では、前に定義した Counter エンティティを再利用します。実際のアプリケーションでは、より詳細な BankAccount エンティティを定義することをお勧めします。

// This is a method called by an orchestrator function
public static async Task<bool> TransferFundsAsync(
    string sourceId,
    string destinationId,
    int transferAmount,
    IDurableOrchestrationContext context)
{
    var sourceEntity = new EntityId(nameof(Counter), sourceId);
    var destinationEntity = new EntityId(nameof(Counter), destinationId);

    // Create a critical section to avoid race conditions.
    // No operations can be performed on either the source or
    // destination accounts until the locks are released.
    using (await context.LockAsync(sourceEntity, destinationEntity))
    {
        ICounter sourceProxy =
            context.CreateEntityProxy<ICounter>(sourceEntity);
        ICounter destinationProxy =
            context.CreateEntityProxy<ICounter>(destinationEntity);

        int sourceBalance = await sourceProxy.Get();

        if (sourceBalance >= transferAmount)
        {
            await sourceProxy.Add(-transferAmount);
            await destinationProxy.Add(transferAmount);

            // the transfer succeeded
            return true;
        }
        else
        {
            // the transfer failed due to insufficient funds
            return false;
        }
    }
}

.NETでは、LockAsync は IDisposable を返します。破棄することで、クリティカルセクションが終了します。クリティカルセクションを表すには、 using ブロックと共に使用します。

前の例では、オーケストレーター関数を使用して、送金元エンティティから送金先エンティティに送金を行いました。 LockAsync メソッドは、移行元と移行先の両方のアカウントエンティティをロックしました。このロックにより、オーケストレーションロジックが using ステートメントの末尾にあるクリティカルセクションを終了するまで、他のクライアントがいずれかのアカウントの状態を照会または変更することができませんでした。この動作により、ソースアカウントのオーバードラフトが防止されます。

注

オーケストレーションが正常に終了するか、エラーが発生すると、進行中の重要なセクションはすべて暗黙的に終了し、システムはすべてのロックを解放します。

クリティカルセクションの動作

LockAsync メソッドは、オーケストレーションに重要なセクションを作成します。これらのクリティカルセクションにより、指定されているエンティティのセットに対して他のオーケストレーションにより重複する変更が行われることを防止できます。内部的には、 LockAsync API はエンティティに "ロック" 操作を送信し、各エンティティから "ロック取得" 応答を受け取った後に返します。ロックとロック解除はどちらも、すべてのエンティティでサポートされている組み込み操作です。

他のクライアントは、ロックされている間はエンティティに対して操作を実行できません。この動作により、一度に 1 つのオーケストレーションインスタンスだけがエンティティをロックできるようになります。オーケストレーションによってロックされているエンティティに対して呼び出し元が操作を呼び出そうとすると、その操作は保留中操作キューに配置されます。保持しているオーケストレーションがロックを解除するまで、保留中の操作は処理されません。

注

この動作は、C# の lock ステートメントなど、ほとんどのプログラミング言語で使用される同期プリミティブとは若干異なります。たとえば、C# では、すべてのスレッドが適切な同期を確保するために、 lock ステートメントを使用する必要があります。一方、エンティティでは、すべての呼び出し元が明示的にエンティティをロックする必要はありません。いずれかの呼び出し元によってエンティティがロックされた場合、そのエンティティに対する他のすべての操作はブロックされ、キューでそのロックの後に配置されます。

エンティティのロックは持続的であるため、実行プロセスがリサイクルされた場合でも永続化されます。システムは、エンティティの永続的な状態の一部としてロックを保持します。

トランザクションとは異なり、クリティカルセクションでは、エラーが発生した場合に変更が自動的にロールバックされません。代わりに、ロールバックや再試行などのエラー処理を書き込みます。たとえば、エラーや例外をキャッチします。この設計の選択は意図的なものです。一般に、オーケストレーションのすべての効果を自動的にロールバックすることは困難であるか不可能です。これは、オーケストレーションによってアクティビティが実行され、ロールバックできない外部サービスへの呼び出しが行われることがあるためです。また、ロールバックの試み自体が失敗し、さらにエラー処理が必要になる場合があります。

クリティカルセクションのルール

ほとんどのプログラミング言語の低レベルのロックプリミティブとは異なり、クリティカルセクションは デッドロックしないことが保証されています。デッドロックを防ぐために、次の制限が適用されます。

クリティカルセクションを入れ子にすることはできません。
クリティカルセクションでサブオーケストレーションを作成することはできません。
クリティカルセクションでは、ロックするエンティティのみを呼び出すことができます。
クリティカルセクションでは、複数の並列呼び出しを使用して、同じエンティティを呼び出すことはできません。
クリティカルセクションでは、ロックセット外のエンティティのみを通知できます。

これらの規則に違反すると、.NETの LockingRulesViolationException などの実行時エラーが発生します。これには、破損したルールを説明するメッセージが含まれます。