Creación de una Amazon SWF aplicación 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 pom.xml archivo del proyecto y añada el aws-java-sdk-simpleworkflowmódulo añadiendo una dependencia para él dentro del <dependencies> bloque.

<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 1.7+. 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 import declaraciones para la AmazonSimpleWorkflowClientclase y para varias clases en el 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.

Tras las declaraciones de cadena, crea una instancia de la AmazonSimpleWorkflowClientclase. Esta es la interfaz básica de los Amazon SWF métodos proporcionados por AWS SDK para 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 utiliza un perfil diferente, modifique el código anterior de la siguiente manera y profile_name sustitúyalo 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!");
    }

Al registrar un dominio, se le proporciona un nombre (cualquier conjunto de 1 a 256 caracteres excepto:,,, /|, los caracteres de control o la cadena literal «arn») y un período de retención, que es el número de días que se Amazon SWF conservarán los datos del historial de ejecución del flujo de trabajo una vez finalizada 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. Para obtener más información, consulta RegisterDomainRequest.

Si ya existe un dominio con ese nombre, DomainAlreadyExistsExceptionse genera un. 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. Para obtener más información, consulta RegisterActivityTypeRequest.

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 ActivityWorker clase al archivo y asígnele un miembro de datos para que contenga un cliente SWF con Amazon SWF el que interactuaremos:

    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 las tareas Amazon SWF llamando al pollForActivityTask método del cliente SWF y especificando el dominio y la lista de tareas que se van a utilizar en la transferencia. PollForActivityTaskRequest

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 realizó correctamente (no se generó ningún error), el trabajador responde al SWF llamando al respondActivityTaskCompleted método del cliente SWF con un RespondActivityTaskCompletedRequestobjeto que contiene el token de la tarea y los datos de los resultados de la actividad.

Por otro lado, si la tarea ha fallado, respondemos llamando al respondActivityTaskFailed método con un RespondActivityTaskFailedRequestobjeto, pasándole el token de la tarea y la 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 WorkflowWorker clase y cree una instancia de la AmazonSimpleWorkflowClientclase utilizada para acceder a los métodos 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. PollForDecisionTaskRequestProporciona 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 lo esnull, lo procesamos más adelante en el executeDecisionTask método y le pasamos el token de la tarea y la lista de HistoryEventobjetos 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 HistoryEventde la clase para obtener información sobre cada tipo de evento.

+ NOTA: las cadenas de switch se introdujeron en Java 7. Si utilizas una versión anterior de Java, puedes utilizar la EventTypeclase para convertir lo String devuelto por history_event.getType() en un valor de enumeración y, String si es necesario, volver a convertirlo en un valor:

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 de cliente SWF toma un StartWorkflowExecutionRequestobjeto como entrada. startWorkflowExecution Aquí, además de especificar el dominio y tipo de flujo de trabajo que se va a ejecutar, proporcionamos:

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

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