Building a Simple Amazon SWF Application

Start a new project with Maven:

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

This will create a new project with a standard maven project structure:

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

You can ignore or delete the test directory and all it contains, we won’t be using it for this tutorial. You can also delete App.java, since we’ll be replacing it with new classes.

Edit the project’s pom.xml file and add the aws-java-sdk-simpleworkflow module by adding a dependency for it within the <dependencies> block.

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

Make sure that Maven builds your project with JDK 1.7+ support. Add the following to your project (either before or after the <dependencies> block) in pom.xml:

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

Create the file in the in the project’s src/main/java/aws/example/helloswf/ directory.

Add a package declaration to the beginning of each file to declare its namespace. The example project uses:

package aws.example.helloswf;

Add import declarations for the AmazonSimpleWorkflowClient class and for multiple classes in the com.amazonaws.services.simpleworkflow.model namespace. To simplify things, we’ll use:

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

Open your text editor and create the file HelloTypes.java, adding a package declaration and imports according to the common steps.

Declare the HelloTypes class and provide it with values to use for your registered activity and workflow types:

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

These values will be used throughout the code.

After the String declarations, create an instance of the AmazonSimpleWorkflowClient class. This is the basic interface to the Amazon SWF methods provided by the AWS SDK for Java.

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

The previous snippet assumes that temporary credentials are associated with the default profile. If you use a different profile, modify the code above as follows and replace profile_name with the name of actual profile name.

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

Add a new function to register a SWF domain. A domain is a logical container for a number of related SWF activity and workflow types. SWF components can only communicate with each other if they exist within the same domain.

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

When you register a domain, you provide it with a name (any set of 1 - 256 characters excluding :, /, |, control characters or the literal string '`arn') and a retention period, which is the number of days that Amazon SWF will keep your workflow’s execution history data after a workflow execution has completed. The maximum workflow execution retention period is 90 days. See RegisterDomainRequest for more information.

If a domain with that name already exists, a DomainAlreadyExistsException is raised. Because we’re unconcerned if the domain has already been created, we can ignore the exception.

Add a function to register a new activity type. An activity represents a unit of work in your workflow.

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

An activity type is identified by a name and a version, which are used to uniquely identify the activity from any others in the domain that it’s registered in. Activities also contain a number of optional parameters, such as the default task-list used to receive tasks and data from SWF and a number of different timeouts that you can use to place constraints upon how long different parts of the activity execution can take. See RegisterActivityTypeRequest for more information.

Open your text editor and create the file ActivityWorker.java, adding a package declaration and imports according to the common steps.

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

Add the ActivityWorker class to the file, and give it a data member to hold a SWF client that we’ll use to interact with Amazon SWF:

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

Add the method that we’ll use as an activity:

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

The activity simply takes a string, combines it into a greeting and returns the result. Although there is little chance that this activity will raise an exception, it’s a good idea to design activities that can raise an error if something goes wrong.

Add a main method that we’ll use as the activity task polling method. We’ll start it by adding some code to poll the task list for activity tasks:

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

The activity receives tasks from Amazon SWF by calling the SWF client’s pollForActivityTask method, specifying the domain and task list to use in the passed-in PollForActivityTaskRequest.

Once a task is received, we retrieve a unique identifier for it by calling the task’s getTaskToken method.

Next, write some code to process the tasks that come in. Add the following to your main method, right after the code that polls for the task and retrieves its task token.

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

If the task token is not null, then we can start running the activity method (sayHello), providing it with the input data that was sent with the task.

If the task succeeded (no error was generated), then the worker responds to SWF by calling the SWF client’s respondActivityTaskCompleted method with a RespondActivityTaskCompletedRequest object containing the task token and the activity’s result data.

On the other hand, if the task failed, then we respond by calling the respondActivityTaskFailed method with a RespondActivityTaskFailedRequest object, passing it the task token and information about the error.

Open your text editor and create the file WorkflowWorker.java, adding a package declaration and imports according to the common steps.

Add a few additional imports to the file:

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;

Declare the WorkflowWorker class, and create an instance of the AmazonSimpleWorkflowClient class used to access SWF methods.

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

Add the main method. The method loops continuously, polling for decision tasks using the SWF client’s pollForDecisionTask method. The PollForDecisionTaskRequest provides the details.

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

Once a task is received, we call its getTaskToken method, which returns a string that can be used to identify the task. If the returned token is not null, then we process it further in the executeDecisionTask method, passing it the task token and the list of HistoryEvent objects sent with the task.

Add the executeDecisionTask method, taking the task token (a String) and the HistoryEvent list.

    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;

We also set up some data members to keep track of things such as:

Next, add some code to executeDecisionTask to process the HistoryEvent objects that were sent with the task, based on the event type reported by the getEventType method.

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

For the purposes of our workflow, we are most interested in:

The other event types can be used if your workflow requires them. See the HistoryEvent class description for information about each event type.

+ NOTE: Strings in switch statements were introduced in Java 7. If you’re using an earlier version of Java, you can make use of the EventType class to convert the String returned by history_event.getType() to an enum value and then back to a String if necessary:

After the switch statement, add more code to respond with an appropriate decision based on the task that was received.

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

In either case, the decision information is added to the Decision list that was declared at the top of the method.

Complete the decision task by returning the list of Decision objects collected while processing the task. Add this code at the end of the executeDecisionTask method that we’ve been writing:

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

The SWF client’s respondDecisionTaskCompleted method takes the task token that identifies the task as well as the list of Decision objects.

Open your text editor and create the file WorkflowStarter.java, adding a package declaration and imports according to the common steps.

Add the WorkflowStarter class:

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

The WorkflowStarter class consists of a single method, main, which takes an optional argument passed on the command-line as input data for the workflow.

The SWF client method, startWorkflowExecution, takes a StartWorkflowExecutionRequest object as input. Here, in addition to specifying the domain and workflow type to run, we provide it with:

The Run object that startWorkflowExecution returns provides a run ID, a value that can be used to identify this particular workflow execution in Amazon SWF's history of your workflow executions.

+ NOTE: The run ID is generated by Amazon SWF, and is not the same as the workflow execution name that you pass in when starting the workflow execution.