Adding FlowDefinition to PutIntegration

aws-customerprofiles-integration/pom.xml

@@ -22,7 +22,7 @@
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>customerprofiles</artifactId>
-            <version>2.15.38</version>
+            <version>2.16.44</version>
         </dependency>
         <!-- https://mvnrepository.com/artifact/software.amazon.cloudformation/aws-cloudformation-rpdk-java-plugin -->
         <dependency>

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/CreateHandler.java

@@ -2,24 +2,29 @@
 
 import lombok.NoArgsConstructor;
 import software.amazon.awssdk.services.customerprofiles.CustomerProfilesClient;
+import software.amazon.awssdk.services.customerprofiles.model.AccessDeniedException;
 import software.amazon.awssdk.services.customerprofiles.model.BadRequestException;
 import software.amazon.awssdk.services.customerprofiles.model.GetIntegrationRequest;
 import software.amazon.awssdk.services.customerprofiles.model.GetIntegrationResponse;
 import software.amazon.awssdk.services.customerprofiles.model.InternalServerException;
 import software.amazon.awssdk.services.customerprofiles.model.PutIntegrationRequest;
 import software.amazon.awssdk.services.customerprofiles.model.PutIntegrationResponse;
 import software.amazon.awssdk.services.customerprofiles.model.ResourceNotFoundException;
+import software.amazon.awssdk.services.customerprofiles.model.ThrottlingException;
+import software.amazon.cloudformation.exceptions.CfnAccessDeniedException;
 import software.amazon.cloudformation.exceptions.CfnAlreadyExistsException;
 import software.amazon.cloudformation.exceptions.CfnGeneralServiceException;
 import software.amazon.cloudformation.exceptions.CfnInvalidRequestException;
 import software.amazon.cloudformation.exceptions.CfnNotFoundException;
 import software.amazon.cloudformation.exceptions.CfnServiceInternalErrorException;
+import software.amazon.cloudformation.exceptions.CfnThrottlingException;
 import software.amazon.cloudformation.proxy.AmazonWebServicesClientProxy;
 import software.amazon.cloudformation.proxy.Logger;
 import software.amazon.cloudformation.proxy.ProgressEvent;
 import software.amazon.cloudformation.proxy.ResourceHandlerRequest;
 
 import java.util.Map;
+import static software.amazon.customerprofiles.integration.Translator.buildServiceFlowDefinition;
 
 @NoArgsConstructor
 public class CreateHandler extends BaseHandler<CallbackContext> {
@@ -44,6 +49,11 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
 
         final ResourceModel model = request.getDesiredResourceState();
 
+        // calls to GetIntegration without a URI result in a 400 so we can skip calling
+        if (model.getUri() == null) {
+            return createIntegration(proxy, request, logger);
+        }
+
         final GetIntegrationRequest getIntegrationRequest = GetIntegrationRequest.builder()
                 .domainName(model.getDomainName())
                 .uri(model.getUri())
@@ -56,46 +66,7 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
             // 1. BadRequestException will also handled by PutIntegration
             // 2. ResourceNotFoundException is the exact exception we want before calling PutIntegration
             // 3. Whatever 5xx error GetIntegration call meet, it should not affect the performance of Create Action
-
-            Map<String, String> resourceTag;
-            if (request.getDesiredResourceTags() == null) {
-                resourceTag = null;
-            } else if (request.getDesiredResourceTags().isEmpty()) {
-                resourceTag = null;
-            } else {
-                resourceTag = request.getDesiredResourceTags();
-            }
-            final PutIntegrationRequest putIntegrationRequest = PutIntegrationRequest.builder()
-                    .domainName(model.getDomainName())
-                    .objectTypeName(model.getObjectTypeName())
-                    .tags(resourceTag)
-                    .uri(model.getUri())
-                    .build();
-
-            final PutIntegrationResponse putIntegrationResponse;
-            try {
-                putIntegrationResponse = proxy.injectCredentialsAndInvokeV2(putIntegrationRequest, client::putIntegration);
-                logger.log(String.format("Integration Created with domainName = %s", model.getDomainName()));
-            } catch (BadRequestException e) {
-                throw new CfnInvalidRequestException(e);
-            } catch (InternalServerException e) {
-                throw new CfnServiceInternalErrorException(e);
-            } catch (ResourceNotFoundException e) {
-                throw new CfnNotFoundException(e);
-            } catch (Exception e) {
-                throw new CfnGeneralServiceException(e);
-            }
-
-            final ResourceModel responseModel = ResourceModel.builder()
-                    .createdAt(putIntegrationResponse.createdAt().toString())
-                    .domainName(putIntegrationResponse.domainName())
-                    .lastUpdatedAt(putIntegrationResponse.lastUpdatedAt().toString())
-                    .objectTypeName(putIntegrationResponse.objectTypeName())
-                    .tags(Translator.mapTagsToList(putIntegrationResponse.tags()))
-                    .uri(putIntegrationResponse.uri())
-                    .build();
-
-            return ProgressEvent.defaultSuccessHandler(responseModel);
+            return createIntegration(proxy, request, logger);
         }
 
         // If GetIntegration Call succeed
@@ -109,5 +80,65 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
         throw new CfnAlreadyExistsException(e);
     }
 
+    /**
+     * Creates an integration
+     * @param proxy
+     * @param request
+     * @param logger
+     * @return
+     */
+    private ProgressEvent<ResourceModel, CallbackContext> createIntegration(
+            final AmazonWebServicesClientProxy proxy,
+            final ResourceHandlerRequest<ResourceModel> request,
+            final Logger logger
+    ) {
+        final ResourceModel model = request.getDesiredResourceState();
+
+        Map<String, String> resourceTag;
+        if (request.getDesiredResourceTags() == null) {
+            resourceTag = null;
+        } else if (request.getDesiredResourceTags().isEmpty()) {
+            resourceTag = null;
+        } else {
+            resourceTag = request.getDesiredResourceTags();
+        }
+        PutIntegrationRequest putIntegrationRequest = PutIntegrationRequest.builder()
+                .domainName(model.getDomainName())
+                .objectTypeName(model.getObjectTypeName())
+                .tags(resourceTag)
+                .flowDefinition(buildServiceFlowDefinition(model.getFlowDefinition()))
+                .uri(model.getUri())
+                .build();
+
+        final PutIntegrationResponse putIntegrationResponse;
+        try {
+            putIntegrationResponse = proxy.injectCredentialsAndInvokeV2(putIntegrationRequest, client::putIntegration);
+            logger.log(String.format("Integration Created with domainName = %s", model.getDomainName()));
+        } catch (BadRequestException e) {
+            throw new CfnInvalidRequestException(e);
+        } catch (AccessDeniedException e) {
+            throw new CfnAccessDeniedException(e);
+        } catch (ThrottlingException e) {
+            throw new CfnThrottlingException(e);
+        } catch (InternalServerException e) {
+            throw new CfnServiceInternalErrorException(e);
+        } catch (ResourceNotFoundException e) {
+            throw new CfnNotFoundException(e);
+        } catch (Exception e) {
+            throw new CfnGeneralServiceException(e);
+        }
+
+        final ResourceModel responseModel = ResourceModel.builder()
+                .createdAt(putIntegrationResponse.createdAt().toString())
+                .domainName(putIntegrationResponse.domainName())
+                .lastUpdatedAt(putIntegrationResponse.lastUpdatedAt().toString())
+                .objectTypeName(putIntegrationResponse.objectTypeName())
+                .tags(Translator.mapTagsToList(putIntegrationResponse.tags()))
+                .uri(putIntegrationResponse.uri())
+                .build();
+
+        return ProgressEvent.defaultSuccessHandler(responseModel);
+    }
 
 }
+

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/Translator.java

@@ -1,10 +1,21 @@
 package software.amazon.customerprofiles.integration;
 
+import com.amazonaws.util.StringUtils;
+import com.google.common.collect.ImmutableMap;
+import software.amazon.awssdk.services.customerprofiles.model.FlowDefinition;
 import software.amazon.cloudformation.proxy.ResourceHandlerRequest;
 
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import software.amazon.customerprofiles.integration.translators.ConnectorTranslator;
+import software.amazon.customerprofiles.integration.translators.MarketoTranslator;
+import software.amazon.customerprofiles.integration.translators.S3Translator;
+import software.amazon.customerprofiles.integration.translators.SalesforceTranslator;
+import software.amazon.customerprofiles.integration.translators.ServiceNowTranslator;
+import software.amazon.customerprofiles.integration.translators.ZendeskTranslator;
+import static software.amazon.customerprofiles.integration.translators.TaskTranslator.toServiceTasks;
+import static software.amazon.customerprofiles.integration.translators.TriggerConfigTranslator.toServiceTriggerConfig;
 
 public class Translator {
 
@@ -26,4 +37,33 @@ static List<Tag> mapTagsToList(Map<String, String> tags) {
                 .collect(Collectors.toList());
     }
 
+    private static Map<String, ConnectorTranslator> connectorTranslators = ImmutableMap.<String, ConnectorTranslator>builder()
+            .put("Salesforce", new SalesforceTranslator())
+            .put("Marketo", new MarketoTranslator())
+            .put("ServiceNow", new ServiceNowTranslator())
+            .put("S3", new S3Translator())
+            .put("Zendesk", new ZendeskTranslator())
+            .build();
+
+    public static ConnectorTranslator getTranslator(String connectorType) {
+        return connectorTranslators.get(connectorType);
+    }
+
+    public static FlowDefinition buildServiceFlowDefinition(software.amazon.customerprofiles.integration.FlowDefinition model) {
+        if (model == null) {
+            return null;
+        }
+
+        String connectorType = model.getSourceFlowConfig().getConnectorType();
+        ConnectorTranslator connector = getTranslator(connectorType);
+        return FlowDefinition.builder()
+                .description(!StringUtils.isNullOrEmpty(model.getDescription()) ? model.getDescription() : null)
+                .flowName(model.getFlowName())
+                .kmsArn(model.getKmsArn())
+                .sourceFlowConfig(connector.toServiceSourceFlowConfig(model.getSourceFlowConfig()))
+                .tasks(toServiceTasks(model.getTasks(), connectorType))
+                .triggerConfig(toServiceTriggerConfig(model.getTriggerConfig()))
+                .build();
+    }
 }
+

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/translators/ConnectorTranslator.java

@@ -0,0 +1,9 @@
+package software.amazon.customerprofiles.integration.translators;
+
+import software.amazon.awssdk.services.customerprofiles.model.ConnectorOperator;
+import software.amazon.awssdk.services.customerprofiles.model.SourceFlowConfig;
+
+public interface ConnectorTranslator {
+    SourceFlowConfig toServiceSourceFlowConfig(software.amazon.customerprofiles.integration.SourceFlowConfig model);
+    ConnectorOperator toServiceConnectorOperator(software.amazon.customerprofiles.integration.ConnectorOperator model);
+}

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/translators/MarketoTranslator.java

@@ -0,0 +1,44 @@
+package software.amazon.customerprofiles.integration.translators;
+
+import com.amazonaws.util.StringUtils;
+import software.amazon.awssdk.services.customerprofiles.model.ConnectorOperator;
+import software.amazon.awssdk.services.customerprofiles.model.IncrementalPullConfig;
+import software.amazon.awssdk.services.customerprofiles.model.MarketoSourceProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceConnectorProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceFlowConfig;
+import software.amazon.customerprofiles.integration.translators.ConnectorTranslator;
+
+public class MarketoTranslator implements ConnectorTranslator {
+
+    @Override
+    public SourceFlowConfig toServiceSourceFlowConfig(software.amazon.customerprofiles.integration.SourceFlowConfig model) {
+        return SourceFlowConfig.builder()
+                .connectorProfileName(StringUtils.isNullOrEmpty(model.getConnectorProfileName()) ? null :
+                        model.getConnectorProfileName())
+                .connectorType(model.getConnectorType())
+                .sourceConnectorProperties(
+                        SourceConnectorProperties.builder()
+                                .marketo(toServiceSourceProperties(model
+                                        .getSourceConnectorProperties()
+                                        .getMarketo()))
+                                .build()
+                )
+                .build();
+    }
+
+    private MarketoSourceProperties toServiceSourceProperties(software.amazon.customerprofiles.integration.MarketoSourceProperties model) {
+        return MarketoSourceProperties.builder()
+                .object(model.getObject())
+                .build();
+    }
+
+    @Override
+    public ConnectorOperator toServiceConnectorOperator(software.amazon.customerprofiles.integration.ConnectorOperator model) {
+        if (model == null) {
+            return null;
+        }
+        return ConnectorOperator.builder()
+                .marketo(model.getMarketo())
+                .build();
+    }
+}

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/translators/S3Translator.java

@@ -0,0 +1,45 @@
+package software.amazon.customerprofiles.integration.translators;
+
+import com.amazonaws.util.StringUtils;
+import software.amazon.awssdk.services.customerprofiles.model.ConnectorOperator;
+import software.amazon.awssdk.services.customerprofiles.model.IncrementalPullConfig;
+import software.amazon.awssdk.services.customerprofiles.model.S3SourceProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceConnectorProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceFlowConfig;
+import software.amazon.customerprofiles.integration.translators.ConnectorTranslator;
+
+public class S3Translator implements ConnectorTranslator {
+
+    @Override
+    public SourceFlowConfig toServiceSourceFlowConfig(software.amazon.customerprofiles.integration.SourceFlowConfig model) {
+        return SourceFlowConfig.builder()
+                .connectorProfileName(StringUtils.isNullOrEmpty(model.getConnectorProfileName()) ? null :
+                        model.getConnectorProfileName())
+                .connectorType(model.getConnectorType())
+                .sourceConnectorProperties(
+                        SourceConnectorProperties.builder()
+                                .s3(toServiceS3SourceProperties(model
+                                        .getSourceConnectorProperties()
+                                        .getS3()))
+                                .build()
+                )
+                .build();
+    }
+
+    private S3SourceProperties toServiceS3SourceProperties(software.amazon.customerprofiles.integration.S3SourceProperties model) {
+        return S3SourceProperties.builder()
+                .bucketPrefix(model.getBucketPrefix())
+                .bucketName(model.getBucketName())
+                .build();
+    }
+
+    @Override
+    public ConnectorOperator toServiceConnectorOperator(software.amazon.customerprofiles.integration.ConnectorOperator model) {
+        if (model == null) {
+            return null;
+        }
+        return ConnectorOperator.builder()
+                .s3(model.getS3())
+                .build();
+    }
+}

aws-customerprofiles-integration/src/main/java/software/amazon/customerprofiles/integration/translators/SalesforceTranslator.java

@@ -0,0 +1,49 @@
+package software.amazon.customerprofiles.integration.translators;
+
+import com.amazonaws.util.StringUtils;
+import software.amazon.awssdk.services.customerprofiles.model.ConnectorOperator;
+import software.amazon.awssdk.services.customerprofiles.model.IncrementalPullConfig;
+import software.amazon.awssdk.services.customerprofiles.model.SalesforceSourceProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceConnectorProperties;
+import software.amazon.awssdk.services.customerprofiles.model.SourceFlowConfig;
+import software.amazon.customerprofiles.integration.translators.ConnectorTranslator;
+
+public class SalesforceTranslator implements ConnectorTranslator {
+
+    @Override
+    public SourceFlowConfig toServiceSourceFlowConfig(software.amazon.customerprofiles.integration.SourceFlowConfig model) {
+        return SourceFlowConfig.builder()
+                .connectorProfileName(StringUtils.isNullOrEmpty(model.getConnectorProfileName()) ? null :
+                        model.getConnectorProfileName())
+                .connectorType(model.getConnectorType())
+                .incrementalPullConfig(model.getIncrementalPullConfig() == null ? null : IncrementalPullConfig.builder()
+                        .datetimeTypeFieldName(model.getIncrementalPullConfig().getDatetimeTypeFieldName())
+                        .build())
+                .sourceConnectorProperties(
+                        SourceConnectorProperties.builder()
+                                .salesforce(toServiceSourceProperties(model
+                                        .getSourceConnectorProperties()
+                                        .getSalesforce()))
+                                .build()
+                )
+                .build();
+    }
+
+    private SalesforceSourceProperties toServiceSourceProperties(software.amazon.customerprofiles.integration.SalesforceSourceProperties model) {
+        return SalesforceSourceProperties.builder()
+                .enableDynamicFieldUpdate(model.getEnableDynamicFieldUpdate())
+                .includeDeletedRecords(model.getIncludeDeletedRecords())
+                .object(model.getObject())
+                .build();
+    }
+
+    @Override
+    public ConnectorOperator toServiceConnectorOperator(software.amazon.customerprofiles.integration.ConnectorOperator model) {
+        if (model == null) {
+            return null;
+        }
+        return ConnectorOperator.builder()
+                .salesforce(model.getSalesforce())
+                .build();
+    }
+}