シンプルな Amazon SWF アプリケーションの構築

Maven での新しいプロジェクトの開始

mvn archetype:generate -DartifactId=helloswf \
-DgroupId=aws.example.helloswf -DinteractiveMode=false

これにより、標準の Maven プロジェクト構造を持つ新しいプロジェクトが作成されます。

helloswf
├── pom.xml
└── src
    ├── main
    │   └── java
    │       └── aws
    │           └── example
    │               └── helloswf
    │                   └── App.java
    └── test
        └── ...

test ディレクトリとそれに含まれるすべては、このチュートリアルでは使用しないため、無視または削除できます。また、新しいクラスに置き換えるため、App.java も削除できます。

<dependencies> ブロック内で依存関係を追加することで、プロジェクトの pom.xml ファイルを編集して aws-java-sdk-simpleworkflow モジュールを追加します。

<dependencies>
  <dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk-simpleworkflow</artifactId>
    <version>1.11.1000</version>
  </dependency>
</dependencies>

Maven によって JDK 1.7 以降のサポートがあるプロジェクトが構築されることを確認します。プロジェクトの pom.xml に以下を追加します (<dependencies> ブロックの前または後)。

<build>
  <plugins>
    <plugin>
      <artifactId>maven-compiler-plugin</artifactId>
      <version>3.6.1</version>
      <configuration>
          <source>1.8</source>
          <target>1.8</target>
      </configuration>
    </plugin>
  </plugins>
</build>

プロジェクトの src/main/java/aws/example/helloswf/ ディレクトリでファイルを作成します。

各ファイルの先頭に package 宣言を追加して名前空間を宣言します。サンプルプロジェクトでは以下を使用します。

package aws.example.helloswf;

AmazonSimpleWorkflowClient クラスおよび com.amazonaws.services.simpleworkflow.model 名前空間内の複数のクラスの import 宣言を追加します。作業を簡素化するため、以下を使用します。

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

テキストエディターを開き、ファイル HelloTypes.java を作成して、共通のステップに従ってパッケージ宣言とインポートを追加します。

HelloTypes クラスを宣言し、登録されたアクティビティとワークフロータイプで使用する値を指定します。

    public static final String DOMAIN = "HelloDomain";
    public static final String TASKLIST = "HelloTasklist";
    public static final String WORKFLOW = "HelloWorkflow";
    public static final String WORKFLOW_VERSION = "1.0";
    public static final String ACTIVITY = "HelloActivity";
    public static final String ACTIVITY_VERSION = "1.0";

これらの値は、コード全体で使用されます。

文字列を宣言したら、AmazonSimpleWorkflowClient クラスのインスタンスを作成します。これは、AWS SDK for Java で提供される Amazon SWF メソッドの基本インタフェースです。

private static final AmazonSimpleWorkflow swf =
    AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

前述のスニペットでは、一時的な認証情報が default プロファイルに関連付けられていることを前提としています。別のプロファイルを使用する場合は、上記のコードを次のように変更し、profile_name を実際のプロファイル名の名前に置き換えてください。

private static final AmazonSimpleWorkflow swf =
        AmazonSimpleWorkflowClientBuilder
                .standard()
                .withCredentials(new ProfileCredentialsProvider("profile_name"))
                .withRegion(Regions.DEFAULT_REGION)
                .build();

SWF ドメインを登録するための新しい関数を追加します。ドメインは、数多くの関連 SWF アクティビティおよびワークフロータイプ用の論理コンテナです。SWF コンポーネントは、同じドメイン内に存在する場合にのみ相互に通信できます。

    try {
        System.out.println("** Registering the domain '" + DOMAIN + "'.");
        swf.registerDomain(new RegisterDomainRequest()
            .withName(DOMAIN)
            .withWorkflowExecutionRetentionPeriodInDays("1"));
    } catch (DomainAlreadyExistsException e) {
        System.out.println("** Domain already exists!");
    }

ドメインを登録する場合、名前 (:、/、| を除く 1～256 文字、制御文字、またはリテラル文字列「arn」の任意のセット) と保持期間を指定します。保持期間は、ワークフローの実行が完了してから Amazon SWF がワークフローの実行履歴データを保持する日数です。ワークフロー実行の最大保持期間は 90 日です。詳細については、「RegisterDomainRequest」を参照してください。

その名前のドメインがすでに存在する場合、DomainAlreadyExistsException が発生します。ドメインがすでに作成されているかどうかは問題ではないため、この例外は無視できます。

新しいアクティビティタイプを登録する関数を追加します。アクティビティはワークフローのワークの単位を表します。

    try {
        System.out.println("** Registering the activity type '" + ACTIVITY +
            "-" + ACTIVITY_VERSION + "'.");
        swf.registerActivityType(new RegisterActivityTypeRequest()
            .withDomain(DOMAIN)
            .withName(ACTIVITY)
            .withVersion(ACTIVITY_VERSION)
            .withDefaultTaskList(new TaskList().withName(TASKLIST))
            .withDefaultTaskScheduleToStartTimeout("30")
            .withDefaultTaskStartToCloseTimeout("600")
            .withDefaultTaskScheduleToCloseTimeout("630")
            .withDefaultTaskHeartbeatTimeout("10"));
    } catch (TypeAlreadyExistsException e) {
        System.out.println("** Activity type already exists!");
    }

アクティビティタイプは名前およびバージョンによって識別されます。これらは、登録されているドメインでこのアクティビティを他のアクティビティから一意に識別するために使用されます。アクティビティには、SWF からタスクとデータを受け取るために使用されるデフォルトのタスクリスト、アクティビティ実行の各部分にかかる時間に対する制約適用に使用できるさまざまなタイムアウトなど、数多くのオプションパラメーターも含まれます。詳細については、「RegisterActivityTypeRequest」を参照してください。

テキストエディターを開き、ファイル ActivityWorker.java を作成して、共通のステップに従ってパッケージ宣言とインポートを追加します。

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

ActivityWorker クラスをファイルに追加し、Amazon SWF を操作するために使用する SWF クライアントを保持するためのデータメンバーを提供します。

    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

アクティビティとして使用するメソッドを追加します。

private static String sayHello(String input) throws Throwable {
    return "Hello, " + input + "!";
}

アクティビティは単純に文字列を受け取り、それをあいさつに組み合わせて、結果を返します。このアクティビティで例外が発生する可能性はほとんどありませんが、何か問題が発生した場合にエラーを発生させるアクティビティを設計することをお勧めします。

アクティビティタスクのポーリングメソッドとして使用する main メソッドを追加します。最初に、アクティビティタスクについてタスクリストをポーリングするコードを追加します。

        System.out.println("Polling for an activity task from the tasklist '"
                + HelloTypes.TASKLIST + "' in the domain '" +
                HelloTypes.DOMAIN + "'.");

        ActivityTask task = swf.pollForActivityTask(
            new PollForActivityTaskRequest()
                .withDomain(HelloTypes.DOMAIN)
                .withTaskList(
                    new TaskList().withName(HelloTypes.TASKLIST)));

        String task_token = task.getTaskToken();

アクティビティは SWF クライアントの pollForActivityTask メソッドを呼び出して Amazon SWF からタスクを受け取り、渡された PollForActivityTaskRequest で使用するドメインとタスクリストを指定します。

タスクを受け取ったら、タスクの getTaskToken メソッドを呼び出して、その固有の識別子を取得します。

次に、入ってきたタスクを処理するコードを記述します。main メソッドで、タスクをポーリングし、タスクトークンを取得するコードの直後に以下を追加します。

    if (task_token != null) {
        String result = null;
        Throwable error = null;

        try {
            System.out.println("Executing the activity task with input '" +
                    task.getInput() + "'.");
            result = sayHello(task.getInput());
        } catch (Throwable th) {
            error = th;
        }

        if (error == null) {
            System.out.println("The activity task succeeded with result '"
                    + result + "'.");
            swf.respondActivityTaskCompleted(
                new RespondActivityTaskCompletedRequest()
                    .withTaskToken(task_token)
                    .withResult(result));
        } else {
            System.out.println("The activity task failed with the error '"
                    + error.getClass().getSimpleName() + "'.");
            swf.respondActivityTaskFailed(
                new RespondActivityTaskFailedRequest()
                    .withTaskToken(task_token)
                    .withReason(error.getClass().getSimpleName())
                    .withDetails(error.getMessage()));
        }
    }

タスクトークンが null でない場合、アクティビティメソッド (sayHello) の実行を開始し、タスクとともに送信された入力データを指定することができます。

タスクが成功した場合 (エラーが生成されなかった場合)、ワーカーは、タスクトークンとアクティビティの結果データを含む RespondActivityTaskCompletedRequest オブジェクトを使用して SWF クライアントの respondActivityTaskCompleted メソッドを呼び出すことにより、SWF に応答します。

一方、タスクが失敗した場合、respondActivityTaskFailedRespondActivityTaskFailedRequest オブジェクトとともに メソッドを呼び出し、タスクトークンとエラーに関する情報を渡して応答します。

テキストエディターを開き、ファイル WorkflowWorker.java を作成して、共通のステップに従ってパッケージ宣言とインポートを追加します。

いくつかのインポートをファイルに追加します。

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

WorkflowWorker クラスを宣言し、SWF メソッドへのアクセスに使用する AmazonSimpleWorkflowClient クラスのインスタンスを作成します。

    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

main メソッドを追加します。このメソッドは連続してループし、SWF クライアントの pollForDecisionTask メソッドを使用して決定タスクをポーリングします。PollForDecisionTaskRequest によって詳細が提供されます。

    PollForDecisionTaskRequest task_request =
        new PollForDecisionTaskRequest()
            .withDomain(HelloTypes.DOMAIN)
            .withTaskList(new TaskList().withName(HelloTypes.TASKLIST));

    while (true) {
        System.out.println(
                "Polling for a decision task from the tasklist '" +
                HelloTypes.TASKLIST + "' in the domain '" +
                HelloTypes.DOMAIN + "'.");

        DecisionTask task = swf.pollForDecisionTask(task_request);

        String taskToken = task.getTaskToken();
        if (taskToken != null) {
            try {
                executeDecisionTask(taskToken, task.getEvents());
            } catch (Throwable th) {
                th.printStackTrace();
            }
        }
    }

タスクを受け取ったら、その getTaskToken メソッドを呼び出します。これにより、タスクの識別に使用できる文字列が返されます。返されたトークンが null でない場合、executeDecisionTask メソッドでさらに処理し、タスクトークンおよびそのタスクで送信された HistoryEvent オブジェクトのリストを渡します。

executeDecisionTask メソッドを追加し、タスクトークン (String) および HistoryEvent リストを受け取ります。

    List<Decision> decisions = new ArrayList<Decision>();
    String workflow_input = null;
    int scheduled_activities = 0;
    int open_activities = 0;
    boolean activity_completed = false;
    String result = null;

また、以下のような項目を追跡するためにいくつかのデータメンバーを設定します。

次に、executeDecisionTask メソッドによって報告されたイベントタイプに基づいて、タスクとともに送信された HistoryEvent オブジェクトを処理するコードを getEventType に追加します。

System.out.println("Executing the decision task for the history events: [");
for (HistoryEvent event : events) {
    System.out.println("  " + event);
    switch(event.getEventType()) {
        case "WorkflowExecutionStarted":
            workflow_input =
                event.getWorkflowExecutionStartedEventAttributes()
                     .getInput();
            break;
        case "ActivityTaskScheduled":
            scheduled_activities++;
            break;
        case "ScheduleActivityTaskFailed":
            scheduled_activities--;
            break;
        case "ActivityTaskStarted":
            scheduled_activities--;
            open_activities++;
            break;
        case "ActivityTaskCompleted":
            open_activities--;
            activity_completed = true;
            result = event.getActivityTaskCompletedEventAttributes()
                          .getResult();
            break;
        case "ActivityTaskFailed":
            open_activities--;
            break;
        case "ActivityTaskTimedOut":
            open_activities--;
            break;
    }
}
System.out.println("]");

ワークフローでは、以下が最も重要です。

他のイベントタイプは、ワークフローで必要な場合に使用できます。各イベントタイプの詳細については、HistoryEvent クラスの説明を参照してください。

+ 注: switch ステートメントの文字列は Java 7 で導入されました。以前のバージョンの Java を使用している場合は、EventType クラスを使用して、history_event.getType() で返される String を列挙値に変換し、必要に応じて String に戻すことができます。

switch ステートメントの後で、受け取ったタスクに応じて、適切な決定を使用して応答するコードを追加します。

if (activity_completed) {
    decisions.add(
        new Decision()
            .withDecisionType(DecisionType.CompleteWorkflowExecution)
            .withCompleteWorkflowExecutionDecisionAttributes(
                new CompleteWorkflowExecutionDecisionAttributes()
                    .withResult(result)));
} else {
    if (open_activities == 0 && scheduled_activities == 0) {

        ScheduleActivityTaskDecisionAttributes attrs =
            new ScheduleActivityTaskDecisionAttributes()
                .withActivityType(new ActivityType()
                    .withName(HelloTypes.ACTIVITY)
                    .withVersion(HelloTypes.ACTIVITY_VERSION))
                .withActivityId(UUID.randomUUID().toString())
                .withInput(workflow_input);

        decisions.add(
                new Decision()
                    .withDecisionType(DecisionType.ScheduleActivityTask)
                    .withScheduleActivityTaskDecisionAttributes(attrs));
    } else {
        // an instance of HelloActivity is already scheduled or running. Do nothing, another
        // task will be scheduled once the activity completes, fails or times out
    }
}

System.out.println("Exiting the decision task with the decisions " + decisions);

いずれの場合も、メソッドの先頭で Decision リストに決定情報が追加されます。

タスクの処理中に収集された Decision オブジェクトのリストを返して、決定タスクを完了します。このコードを、これまで記述した executeDecisionTask メソッドの最後に追加します。

swf.respondDecisionTaskCompleted(
    new RespondDecisionTaskCompletedRequest()
        .withTaskToken(taskToken)
        .withDecisions(decisions));

SWF クライアントの respondDecisionTaskCompleted メソッドは、タスクと、Decision オブジェクトのリストを識別するタスクトークンを受け取ります。

テキストエディターを開き、ファイル WorkflowStarter.java を作成して、共通のステップに従ってパッケージ宣言とインポートを追加します。

WorkflowStarter クラスを追加します。

package aws.example.helloswf;


import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

public class WorkflowStarter {
    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();
    public static final String WORKFLOW_EXECUTION = "HelloWorldWorkflowExecution";

    public static void main(String[] args) {
        String workflow_input = "{SWF}";
        if (args.length > 0) {
            workflow_input = args[0];
        }

        System.out.println("Starting the workflow execution '" + WORKFLOW_EXECUTION +
                "' with input '" + workflow_input + "'.");

        WorkflowType wf_type = new WorkflowType()
            .withName(HelloTypes.WORKFLOW)
            .withVersion(HelloTypes.WORKFLOW_VERSION);

        Run run = swf.startWorkflowExecution(new StartWorkflowExecutionRequest()
            .withDomain(HelloTypes.DOMAIN)
            .withWorkflowType(wf_type)
            .withWorkflowId(WORKFLOW_EXECUTION)
            .withInput(workflow_input)
            .withExecutionStartToCloseTimeout("90"));

        System.out.println("Workflow execution started with the run id '" +
                run.getRunId() + "'.");
    }
}

WorkflowStarter クラスは単一のメソッド main で構成されます。このメソッドはコマンドラインで渡されたオプションの引数を、ワークフローの入力データとして受け取ります。

SWF のクライアントメソッド startWorkflowExecution は、StartWorkflowExecutionRequest オブジェクトを入力として受け取ります。ここで、ドメインと実行するワークフロータイプの指定に加えて、以下を指定します。

が返す RunstartWorkflowExecution オブジェクトは、実行 ID を提供します。これは、ワークフロー実行の Amazon SWF の履歴でこの特定のワークフロー実行を識別するために使用できる値です。

+ 注: 実行 ID は Amazon SWF によって生成され、ワークフローの実行を開始するときに渡すワークフロー実行名と同じではありません。