Creación de una aplicación de Amazon SWF sencilla

Inicie un nuevo proyecto con Maven:

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

Se creará un nuevo proyecto con una estructura de proyecto de Maven estándar:

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

Puede omitir o eliminar el directorio test y todo su contenido; no lo usaremos en este tutorial. También puede eliminar App.java, ya que lo reemplazaremos por nuevas clases.

Edite el archivo pom.xml del proyecto y añada el módulo aws-java-sdk-simpleworkflow añadiendo una dependencia para él en el bloque <dependencies>.

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

Asegúrese de que Maven compila su proyecto de manera que sea compatible con JDK. Añada lo siguiente a su proyecto (delante o detrás del bloque <dependencies>) en 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>

Cree el archivo en el directorio src/main/java/aws/example/helloswf/ del proyecto.

Añada una declaración package al principio de cada archivo para declarar su espacio de nombres. El proyecto de ejemplo usa:

package aws.example.helloswf;

Añada declaraciones import para la clase AmazonSimpleWorkflowClient y para varias clases del espacio de nombres com.amazonaws.services.simpleworkflow.model. Para simplificar las cosas, vamos a utilizar:

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

Abra su editor de texto y cree el archivo HelloTypes.java, añadiendo una declaración del paquete y funciones import de acuerdo con los pasos comunes.

Declare la clase HelloTypes y proporcione los valores que se van a usar para los tipos de actividad y flujo de trabajo registrados:

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

Estos valores se utilizarán en todo el código.

Detrás de las declaraciones de cadena, cree una instancia de la clase AmazonSimpleWorkflowClient. Esta es la interfaz básica a los métodos de Amazon SWF proporcionados por AWS SDK for Java.

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

En el fragmento anterior se presupone que las credenciales temporales están asociadas al perfil default. Si usa un perfil diferente, modifique el código anterior de la siguiente manera y sustituya profile_name por el nombre del perfil real.

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

Añada una nueva función para registrar un dominio de SWF. Un dominio es un contenedor lógico para una serie de tipos de actividad y flujo de trabajo de SWF relacionados. Los componentes de SWF solo pueden comunicarse entre sí si se encuentran en el mismo dominio.

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

Cuando registra un dominio, proporciona un nombre (cualquier conjunto de 1-256 caracteres excepto :, /, |, caracteres de control o la cadena literal 'arn') y un periodo de retención, que es el número de días que Amazon SWF conservará los datos del historial de ejecución del flujo de trabajo una vez completada la ejecución del flujo de trabajo. El periodo máximo de retención de ejecución del flujo de trabajo es de 90 días. Consulte RegisterDomainRequest para obtener más información.

Si ya existe un dominio con ese nombre, se produce una excepción DomainAlreadyExistsException. Como no nos interesa si se ha creado o no se ha creado el dominio, podemos omitir esta excepción.

Añada una función para registrar un nuevo tipo de actividad. Una actividad representa una unidad de trabajo de su flujo de trabajo.

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

Un tipo de actividad se identifica mediante un nombre y una versión, que se utilizan para identificar de forma inequívoca la actividad de otras actividades que se han registrado en el dominio. Las actividades también contienen una serie de parámetros opcionales, como la lista de tareas predeterminadas para recibir las tareas y los datos de SWF, y una serie de tiempos de espera diferentes que puede utilizar para aplicar restricciones sobre cuánto tiempo pueden tardar las distintas partes de la ejecución de la actividad. Consulte RegisterActivityTypeRequest para obtener más información.

Abra su editor de texto y cree el archivo ActivityWorker.java, añadiendo una declaración del paquete y funciones import de acuerdo con los pasos comunes.

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

Añada la clase ActivityWorker al archivo y proporcione un miembro de datos para almacenar el cliente de SWF que usaremos para interactuar con Amazon SWF:

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

Añada el método que usaremos como actividad:

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

La actividad simplemente toma una cadena, la combina en un saludo y devuelve el resultado. Aunque no es muy probable que esta actividad produzca una excepción, es aconsejable que diseñe actividades que puedan producir un error si algo va mal.

Añada un método main que utilizaremos como el método de sondeo de tareas de actividad. Empezaremos añadiendo código para buscar tareas de actividad en la lista de tareas:

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

La actividad recibe tareas de Amazon SWF llamando al método pollForActivityTask del cliente de SWF, especificando el dominio y la lista de tareas que se van a utilizar en el objeto PollForActivityTaskRequest pasado al método.

Una vez que se recibe una tarea, recuperamos un identificador único llamando al método getTaskToken de la tarea.

A continuación, escribimos código para procesar las tareas que llegan. Añada lo siguiente al método main, justo detrás del código que busca la tarea y recupera su token de tarea.

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

Si el token de tarea no es null, podemos empezar a ejecutar el método de actividad (sayHello), facilitándole los datos de entrada que se enviaron con la tarea.

Si la tarea se ejecuta correctamente (no se genera ningún error), el proceso de trabajo responde a SWF llamando al método respondActivityTaskCompleted del cliente de SWF con un objeto RespondActivityTaskCompletedRequest que contiene el token de la tarea y los datos de los resultados de la actividad.

Sin embargo, si la tarea ha producido un error, respondemos llamando al método respondActivityTaskFailed con un objeto RespondActivityTaskFailedRequest, proporcionándole el token de la tarea e información sobre el error.

Abra su editor de texto y cree el archivo WorkflowWorker.java, añadiendo una declaración del paquete y funciones import de acuerdo con los pasos comunes.

Añada algunas funciones import adicionales al archivo:

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 la clase WorkflowWorker y cree una instancia de la clase AmazonSimpleWorkflowClient utilizada para obtener acceso a los métodos de SWF.

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

Añada el método main. Este método se ejecuta en bucle continuamente en busca de tareas de decisión usando el método pollForDecisionTask del cliente de SWF. El objeto PollForDecisionTaskRequest proporciona los detalles.

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

Una vez que se recibe una tarea, llamamos a su método getTaskToken, que devuelve una cadena que se puede utilizar para identificar la tarea. Si el token devuelto no es null, lo seguimos procesando en el método executeDecisionTask, pasándole el token de la tarea y la lista de objetos HistoryEvent enviados con la tarea.

Añada el método executeDecisionTask, tomando el token de tarea (String) y la lista 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;

También configuramos algunos miembros de datos para realizar un seguimiento de cosas como:

A continuación, añada código a executeDecisionTask para procesar los objetos HistoryEvent que se han enviado a la tarea, en función del tipo de evento notificado por el método 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("]");

A efectos de nuestro flujo de trabajo, lo que más nos interesa es:

Los demás tipos de eventos se pueden utilizar si el flujo de trabajo así lo requiere. Consulte la descripción de la clase HistoryEvent para obtener información acerca de cada tipo de evento.

+ NOTA: las cadenas de switch se introdujeron en Java 7. Si utiliza una versión anterior de Java, puede utilizar la clase EventType para convertir el valor String devuelto por history_event.getType() en un valor enum y luego en un valor String si es necesario:

Detrás de la instrucción switch, añada más código para responder con una decisión adecuada en función de la tarea que se ha recibido.

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

En cualquier caso, la información de la decisión se añade a la lista Decision que se declaró encima del método.

Complete la tarea de decisión devolviendo la lista de objetos Decision recopilados al procesar la tarea. Añada este código al final del método executeDecisionTask que hemos escrito:

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

El método respondDecisionTaskCompleted del cliente de SWF toma el token de la tarea que identifica la tarea, así como la lista de objetos Decision.

Abra su editor de texto y cree el archivo WorkflowStarter.java, añadiendo una declaración del paquete y funciones import de acuerdo con los pasos comunes.

Añada la clase 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() + "'.");
    }
}

La clase WorkflowStarter consta de un único método, mainque toma un argumento opcional pasado en la línea de comandos como datos de entrada para el flujo de trabajo.

El método del cliente de SWF, startWorkflowExecution, toma un objeto StartWorkflowExecutionRequest como entrada. Aquí, además de especificar el dominio y tipo de flujo de trabajo que se va a ejecutar, proporcionamos:

El objeto Run que devuelve startWorkflowExecution proporciona un ID de ejecución, que es un valor que se puede utilizar para identificar esta ejecución del flujo de trabajo en concreto en el historial de ejecuciones del flujo de trabajo de Amazon SWF.

+ NOTA: el ID de ejecución lo genera Amazon SWF y no es el mismo que el nombre de ejecución del flujo de trabajo que se pasa al iniciar la ejecución del flujo de trabajo.