Desligar operadores de atividade e de fluxo de trabalho de maneira tranquila

O tópico Compilar um aplicativo Amazon SWF simples apresentou uma implementação completa de um aplicativo de fluxo de trabalho simples que consiste em um aplicativo de registro, um operador de atividade e de fluxo de trabalho, além de um início de fluxo de trabalho.

As classes de operador foram projetadas para serem executadas continuamente, sondando tarefas enviadas pelo Amazon SWF para executar atividades ou retornar decisões. Assim que uma requisição de sondagem for feita, o Amazon SWF vai registrar quem fez a sondagem e tentar atribuir uma tarefa.

Se o operador de fluxo de trabalho for encerrado durante uma sondagem longa, o Amazon SWF ainda poderá tentar enviar uma tarefa para o operador encerrado, resultando em uma tarefa perdida (até a tarefa expirar).

Uma maneira de processar essa situação é aguardar todas as requisições de sondagem longa retornarem antes do operador terminar.

Neste tópico, vamos reescrever o operador de atividade de helloswf, usando ganchos de desligamento do Java para fazer um desligamento normal do operador de atividade.

Aqui está o código completo:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.ActivityTask;
import com.amazonaws.services.simpleworkflow.model.PollForActivityTaskRequest;
import com.amazonaws.services.simpleworkflow.model.RespondActivityTaskCompletedRequest;
import com.amazonaws.services.simpleworkflow.model.RespondActivityTaskFailedRequest;
import com.amazonaws.services.simpleworkflow.model.TaskList;

public class ActivityWorkerWithGracefulShutdown {

    private static final AmazonSimpleWorkflow swf =
        AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();
    private static final CountDownLatch waitForTermination = new CountDownLatch(1);
    private static volatile boolean terminate = false;

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

    public static void main(String[] args) {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                try {
                    terminate = true;
                    System.out.println("Waiting for the current poll request" +
                            " to return before shutting down.");
                    waitForTermination.await(60, TimeUnit.SECONDS);
                }
                catch (InterruptedException e) {
                    // ignore
                }
            }
        });
        try {
            pollAndExecute();
        }
        finally {
            waitForTermination.countDown();
        }
    }

    public static void pollAndExecute() {
        while (!terminate) {
            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 taskToken = task.getTaskToken();

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

                try {
                    System.out.println("Executing the activity task with input '"
                            + task.getInput() + "'.");
                    result = executeActivityTask(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(taskToken)
                            .withResult(result));
                }
                else {
                    System.out.println("The activity task failed with the error '"
                            + error.getClass().getSimpleName() + "'.");
                    swf.respondActivityTaskFailed(
                        new RespondActivityTaskFailedRequest()
                            .withTaskToken(taskToken)
                            .withReason(error.getClass().getSimpleName())
                            .withDetails(error.getMessage()));
                }
            }
        }
    }
}

Nesta versão, o código de sondagem que estava na função main na versão original foi migrado para o próprio método, pollAndExecute.

Agora a função main usa um CountDownLatch com um hook de desligamento para fazer o thread aguardar até 60 segundos após o término ser solicitado antes de permitir o desligamento do thread.