aws-step-functions-basics.template

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Description": "An example template with an IAM role for a Lambda state machine.",
  "Resources": {
    "LambdaExecutionRole": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": {
                "Service": "lambda.amazonaws.com"
              },
              "Action": "sts:AssumeRole"
            }
          ]
        }
      }
    },
    "LambdaFunction": {
      "Type": "AWS::Lambda::Function",
      "Properties": {
        "Handler": "index.handler",
        "Role": {
          "Fn::GetAtt": [
            "LambdaExecutionRole",
            "Arn"
          ]
        },
        "Code": {
          "ZipFile": "exports.handler = (event, context, callback) => {\n    callback(null, {value: event.value});\n};\n"
        },
        "Runtime": "nodejs12.x",
        "Timeout": "25"
      }
    },
    "StatesExecutionRole": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": {
                "Service": [
                  {
                    "Fn::Sub": "states.${AWS::Region}.amazonaws.com"
                  }
                ]
              },
              "Action": "sts:AssumeRole"
            }
          ]
        },
        "Path": "/",
        "Policies": [
          {
            "PolicyName": "StatesExecutionPolicy",
            "PolicyDocument": {
              "Version": "2012-10-17",
              "Statement": [
                {
                  "Effect": "Allow",
                  "Action": [
                    "lambda:InvokeFunction"
                  ],
                  "Resource": "*"
                }
              ]
            }
          }
        ]
      }
    },
    "CustomActivity" : {
      "Type" : "AWS::StepFunctions::Activity",
      "Properties" : {
        "Name" : "customActivity"
      }
    },
    "StateMachine": {
      "Type": "AWS::StepFunctions::StateMachine",
      "Properties": {
        "DefinitionString": {
          "Fn::Sub": [
            "{\n  \"Comment\": \"Task state with Lambda and local Activity passed through a decision maker.\",\n  \"StartAt\": \"ChoiceState\",\n  \"Version\": \"1.0\",\n  \"TimeoutSeconds\": 300,\n  \"States\": {\n    \"getLocalState\": {\n      \"Type\": \"Task\",\n      \"Resource\": \"${activityArn}\",\n      \"End\": true\n    },\n    \"getServerlessState\": {\n      \"Type\": \"Task\",\n      \"Resource\": \"${lambdaArn}\",\n      \"Next\": \"getLocalState\"\n    },\n    \"ChoiceState\": {\n      \"Type\": \"Choice\",\n      \"Choices\": [\n        {\n          \"Not\": {\n            \"Variable\": \"$.type\",\n            \"StringEquals\": \"private\"\n          },\n          \"Next\": \"getServerlessState\"\n        },\n        {\n          \"Variable\": \"$.value\",\n          \"NumericEquals\": 0,\n          \"Next\": \"getLocalState\"\n        }\n      ],\n      \"Default\": \"DefaultState\"\n    },\n    \"DefaultState\": {\n      \"Type\": \"Fail\",\n      \"Cause\": \"Value provided does not cover our cases (private & 0)!\"\n    }\n  }\n}",
            {
              "activityArn": { "Ref" : "CustomActivity" },
              "lambdaArn": {
                "Fn::GetAtt": [
                  "LambdaFunction",
                  "Arn"
                ]
              }
            }
          ]
        },
        "RoleArn": {
          "Fn::GetAtt": [
            "StatesExecutionRole",
            "Arn"
          ]
        }
      }
    }
  }
}

LocalCode.java

package main;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.stepfunctions.AWSStepFunctions;
import com.amazonaws.services.stepfunctions.AWSStepFunctionsClientBuilder;
import com.amazonaws.services.stepfunctions.model.GetActivityTaskRequest;
import com.amazonaws.services.stepfunctions.model.GetActivityTaskResult;
import com.amazonaws.services.stepfunctions.model.SendTaskFailureRequest;
import com.amazonaws.services.stepfunctions.model.SendTaskSuccessRequest;
import com.amazonaws.util.json.Jackson;
import com.fasterxml.jackson.databind.JsonNode;

import java.util.concurrent.TimeUnit;

public class LocalCode {

    public static final String ACTIVITY_ARN = "TO_BE_ADDED";

    public String getLocalCode(Integer value) {
        System.out.println("Received payload with number: " + value);
        return "{\"Code\": \"" + value + "\"}";
    }

    public static void main(final String[] args) {
        LocalCode code = new LocalCode();
        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientConfiguration.setSocketTimeout((int) TimeUnit.SECONDS.toMillis(70));

        AWSStepFunctions client = AWSStepFunctionsClientBuilder.standard()
                .withRegion(Regions.EU_CENTRAL_1)
                .withCredentials(new ProfileCredentialsProvider())
                .withClientConfiguration(clientConfiguration)
                .build();

        System.out.println("STARTED listening for the Activity payload!");
        try {
            while (true) {
                GetActivityTaskResult getActivityTaskResult =
                        client.getActivityTask(new GetActivityTaskRequest().withActivityArn(ACTIVITY_ARN));

                if (getActivityTaskResult.getTaskToken() != null) {
                    System.out.println("Found result from previous step.");
                    try {
                        JsonNode json = Jackson.jsonNodeOf(getActivityTaskResult.getInput());
                        String result = code.getLocalCode(json.get("value").intValue());
                        client.sendTaskSuccess(
                                new SendTaskSuccessRequest().withOutput(
                                        result).withTaskToken(getActivityTaskResult.getTaskToken()));
                    } catch (Exception e) {
                        client.sendTaskFailure(new SendTaskFailureRequest().withTaskToken(getActivityTaskResult.getTaskToken()));
                    }
                } else {
                    Thread.sleep(1000);
                }
            }
        } catch (Exception e) {
            System.out.println("EXCEPTION while listening for the Activity payload!");
            e.printStackTrace();
        } finally {
            System.out.println("STOPPED listening for the Activity payload!");
        }
    }
}