Customization Guidelines for Connect SDK

1@HandlerInfo(name = "CustomCategoryUpsert", priority = 100)

1{
2  "configuration": {
3    "active": [],
4    "inactive": []
5  },
6  "job": {
7    "active": [
8      {
9        "route": "event-failure-summary-monitor",
10        "handler": "EventFailureSummaryJobHandler",
11        "priority": 0,
12        "props": {
13          "page-size": "100",
14          "event-status": "FAILED",
15          "event-type": "ORCHESTRATION_AUDIT"
16        },
17        "className": "com.fluentcommerce.connect.core.job.handler.EventFailureSummaryJobHandler",
18        "description": "Grabs failed events from Fluent and builds a summary"
19      }
20    ],
21    "inactive": []
22  },
23  "message": {
24    "active": [],
25    "inactive": []
26  },
27  "notification": {
28    "active": [
29      {
30        "route": "notification",
31        "handler": "DefaultNotificationHandler",
32        "priority": 0,
33        "props": {},
34        "className": "com.fluentcommerce.connect.core.handler.DefaultNotificationHandler",
35        "description": "Handles notifications and publishes to their designated route"
36      }
37    ],
38    "inactive": []
39  }
40}

1{
2  "configuration": {
3    "active": [
4      {
5        "name": "log-level-update",
6        "route": "",
7        "priority": 0,
8        "props": {},
9        "className": "com.fluentcommerce.connect.core.configuration.handler.LogLevelConfigurationHandler"
10      }
11    ],
12    "inactive": []
13  },
14  "job": {
15    "active": [
16      {
17        "name": "EventFailureSummaryJobHandler",
18        "route": "event-failure-summary-monitor",
19        "priority": 0,
20        "props": {
21          "page-size": "100",
22          "event-status": "FAILED",
23          "event-type": "ORCHESTRATION_AUDIT"
24        },
25        "className": "com.fluentcommerce.connect.core.job.handler.EventFailureSummaryJobHandler"
26      }
27    ],
28    "inactive": []
29  },
30  "message": {
31    "active": [],
32    "inactive": []
33  },
34  "notification": {
35    "active": [
36      {
37        "name": "DefaultNotificationHandler",
38        "route": "notification",
39        "priority": 0,
40        "props": {},
41        "className": "com.fluentcommerce.connect.core.handler.DefaultNotificationHandler"
42      }
43    ],
44    "inactive": []
45  }
46}

1@Slf4j
2@Component
3@HandlerInfo(
4        name = "ProductUpsert",
5        route = "commercetools.connect.product.upsert",
6        description = "Updates products in Fluent.",
7        props = {@HandlerProp(key = "query", value = "ct-product.graphql")}
8)
9public class ProductUpsertHandler implements MessageHandler {
10
11    @Override
12    public void processMessage(final MessageHandlerContext context) throws UnprocessableMessageException, HandlerException {
13        final Optional<MyPayload> payload = context.getMessagePayload(MyPayload.class);
14        if (payload.isPresent()) {
15            //logic to process a message
16            final Event productEvent = processProduct(payload);
17            //send product to Fluent
18            context.ofFluentContext().sendEvent(productEvent);
19        }
20        throw new UnprocessableMessageException("Unable to get message payload");      
21    }
22}

1# Route mappings are used to determine a route to a handler when receiving external messages
2route-mapping:
3- route: "commercetools.connect.product.upsert"
4    props:
5    # any number of properties that can be used to determine if this route is adequate for the message received
6        name: "product"
7        inclusion-filter:
8        - "ResourceUpdated"
9        - "ResourceCreated"
10
11# Handler routes          
12routes:
13message:
14    -   route: "commercetools.connect.product.upsert"
15        handler-name: "ProductUpsert"
16        props:
17        # any number of properties can drive or assist the handler to process a message
18            query: "ct-product.graphql"

1@Slf4j
2@Component
3@HandlerInfo(
4        name = "BatchInventorySyncJob",
5        route = "batch-inventory-sync",
6        description = "Sync inventory details from Fluent to commerceTools",
7        props = {@HandlerProp(key = "page-size", value = "500")}
8)
9public class InventoryJobHandler implements JobHandler {
10
11    @Override
12    public void run(final JobHandlerContext jobHandlerContext) throws JobExecutionException {
13        final JobProperties settings = jobHandlerContext.getJobSettings();
14        final LocalDateTime start = settings.getStart();
15        final LocalDateTime end = settings.getEnd();
16
17        //get a runtime configuration propperty 
18        final String myProp = settings.getProp("my-property", StringUtils.EMPTY);
19
20        //get a build time (route) configuration property
21        final int pageSize = jobHandlerContext.getRouteSettings().getProp("page-size", 100);
22
23        //your job logic next...
24
25        //poll external system
26        //either add each item into the queue for async processing 
27            //jobHandlerContext.getPublishService().publishMessage(sdkMessage);
28        //or process each received item here
29    }
30}

1    routes:
2    job:
3        -   route: "fc.connect.batch-inventory-sync"
4            handler-name: "BatchInventorySyncJob"
5            props:
6            page-size: 500

1@Slf4j
2@Component
3@HandlerInfo(name = "log-level-update", route="")
4public class LogLevelConfigurationHandler implements ConfigurationHandler {
5    private static final String DEFAULT_LEVEL = "INFO";
6    private final FluentConnectConfiguration connectConfiguration;
7
8    private Level currentLevel;
9
10    @Autowired
11    public LogLevelConfigurationHandler(final FluentConnectConfiguration connectConfiguration) {
12        this.connectConfiguration = connectConfiguration;
13    }
14
15    @Override
16    public boolean validate(@NotNull final Configuration configuration) {
17        final var newLevel = getLogLevel(configuration);
18
19        final var shouldExecute = currentLevel != null && newLevel.levelInt != currentLevel.levelInt;
20        currentLevel = newLevel;
21        return shouldExecute;
22    }
23
24    @Override
25    public void run(final @NotNull Configuration configuration) {
26        final LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
27        final Logger logger = loggerContext.getLogger("com.fluentcommerce.connect");
28        if (logger != null) {
29            logger.setLevel(getLogLevel(configuration));
30            log.info("Log level for com.fluentcommerce.connect set to [{}].", logger.getLevel());
31
32        } else {
33            log.error("Unable to detect logger context: com.fluentcommerce.connect.");
34        }
35    }
36
37    private Level getLogLevel(@NotNull final Configuration configuration) {
38        final String key = connectConfiguration.getConfiguration().getLogLevelKey();
39        final ConfigurationTuple newValue = configuration.getProperties().getOrDefault(key, ConfigurationTuple.ofReadOnly(key, DEFAULT_LEVEL));
40
41        return StringUtils.isNotBlank(newValue.getValue()) ? Level.toLevel(newValue.getValue()) : Level.INFO;
42    }
43}

1<dependency>
2    <groupId>com.fluentcommerce.connect</groupId>
3    <artifactId>connect-sdk-core-web-security</artifactId>
4</dependency>

1security:
2    web:
3      publicEndpoints:
4        - "/actuator/health"
5        - "/api/docs"
6        - "/swagger-ui.html"
7        - "/api/v1/fluent-connect/webhook"
8      customAuthFilters:
9        - "com.fluentcommerce.connect.core.web.security.filters.ApiKeyAuthenticationFilter"
10        - "com.fluentcommerce.connect.core.web.security.filters.FluentOauthTokenAuthenticationFilter"

1@SdkAuthenticationFilterInfo(provider = "api-key-provider", converter = "api-key-converter")
2public class ApiKeyAuthenticationFilter extends SdkAuthenticationFilter {
3
4    public ApiKeyAuthenticationFilter(
5        @NotNull final AuthenticationManager authenticationManager,
6        @NotNull final AuthenticationConverter authenticationConverter) {
7
8        super(authenticationManager, authenticationConverter);
9    }
10}

1@Slf4j
2@Component("api-key-converter")
3public class ApiKeyDetailAuthenticationConverter implements AuthenticationConverter {
4    protected static final String FLUENT_ACCOUNT_HEADER = "fluent.account";
5    private static final String SECURITY_TOKEN_GROUP = "securityToken";
6
7    private static final String BEARER_TOKEN_PATTERN = "ApiKey\\s(?<securityToken>.*)";
8    @Value("${fluent-connect.mock.accountId:}")
9    protected String fluentAccount;
10
11    @Override
12    public Authentication convert(final HttpServletRequest request) {
13        if (log.isTraceEnabled()) {
14            log.trace("Authentication filter running for request URL: {}", request.getRequestURL().toString());
15        }
16
17        try {
18            final var authApiKey = resolveSecurityTokenFromHttpRequest(request);
19            return new ApiKeyDetailAuthentication(authApiKey, getAccountFromRequest(request));
20        } catch (final BadCredentialsException e) {
21            //returning null allows the spring security to try the next security filter
22            return null;
23        }
24    } 
25    ...
26}

1@Slf4j
2@Component("api-key-provider")
3public class ApiKeyDetailAuthenticationProvider implements AuthenticationProvider {
4    private final APIKeyService apiKeyService;
5    private final ContextBuilder contextBuilder;
6
7    public ApiKeyDetailAuthenticationProvider(final APIKeyService apiKeyService, final ContextBuilder contextBuilder) {
8        this.apiKeyService = apiKeyService;
9        this.contextBuilder = contextBuilder;
10    }
11
12    @Override
13    public Authentication authenticate(final Authentication authentication) throws AuthenticationException {
14        final var authApiKey = (String) authentication.getCredentials();
15        final var accountId = (String) authentication.getPrincipal();
16
17        final var keyDetails = validateApiKeyDetails(authApiKey, accountId);
18        ((ApiKeyDetailAuthentication) authentication).init(keyDetails, contextBuilder);
19        authentication.setAuthenticated(true);
20        return authentication;
21    }
22    ...
23}    

1@Slf4j
2@RestController
3@RequestMapping("/sample")
4public class SampleController {
5    /**
6     * SDK service that has access to the spring security context and returns a SDK context based on the authenticated/authorised credentials
7     */
8    private final SdkSecurityService securityService;
9
10    @Autowired
11    public SampleController(final SdkSecurityService securityService) {
12        this.securityService = securityService;
13    }
14
15    @RequestMapping(method = RequestMethod.POST, consumes = "application/json")
16    public ResponseEntity<String> receive(@RequestBody final SampleRequest sampleRequest) {
17        return securityService.getAuthenticatedContext()
18            .map(context -> processRequest(context, sampleRequest))
19            .orElseThrow(() -> new BadCredentialsException("Unable to authenticate request"));
20    }
21
22    private ResponseEntity<String> processRequest(final HandlerContext handlerContext, final SampleRequest sampleRequest) {
23        //TODO your logic / call a service 
24        return ResponseEntity.ok().build();
25    }
26
27    @Value
28    public static class SampleRequest {
29        String id;
30        String name;
31        //...
32    }
33}

1@Slf4j
2@Component
3@ListenerInfo(id = "my_queue_id", messageClass = SampleQueueListener.MyMessagePayload.class)
4public class SampleQueueListener extends ExternalListener<SampleQueueListener.MyMessagePayload> {
5
6  @Override
7  public Optional<ConnectSDKMessage> receiveMessage(final ExternalMessage message) throws UnprocessableMessageException {
8      final MyMessagePayload messagePayload = getMessageContent(message);
9
10      final Optional<AccountReference> accountReference = getFluentAccountReference(Collections.singletonMap("website-id", messagePayload.websiteId));
11      if (accountReference.isPresent()) {
12          return Optional.ofNullable(getConnectSDKMessage(messagePayload, accountReference.get()));
13      }
14
15      log.error("Unable to retrieve a Fluent account for the message received [{}].", messagePayload.getId());
16      throw new UnprocessableMessageException("Unable to retrieve a Fluent account for the message received.");
17  }
18
19  /**
20   * Extract a subset of the payload, pre-process and enrich it, convert into a different format or simply use it as is.
21   */
22  private ConnectSDKMessage getConnectSDKMessage(final MyMessagePayload messagePayload,
23                                                 final AccountReference accountReference) throws UnprocessableMessageException {
24      //The example below is a pass-through
25      try {
26          return ConnectSDKMessage.builder()
27                  .id(UUID.randomUUID().toString())
28                  .name(getMessageRoute(messagePayload))
29                  .accountId(accountReference.getAccountId())
30                  .retailerId(accountReference.getRetailerId())
31                  .payload(ConnectSDKMessage.toJson(messagePayload))
32                  .build();
33      } catch (JsonProcessingException e) {
34          throw new UnprocessableMessageException(e);
35      }
36  }
37
38  /**
39   * Apply your logic to determine the route name
40   */
41  private String getMessageRoute(final MyMessagePayload messagePayload) {
42      return "my-route";
43  }
44
45  @Value
46  public static class MyMessagePayload {
47      String id;
48      String websiteId;
49      //... 
50  }
51}

1my_queue_id:
2    name: "SQS_CONNECTOR"
3    fluent-account: "DEV_ACCOUNT"
4    type: sqs

1/**
2 * Implementation of a secrets manager service using AWS Secrets Manager
3 */
4@Slf4j
5@Service
6@CredentialManagerInfo(id = "aws-secrets-manager")
7public class AWSCredentialService implements SecretManagerService {
8    public static final String VERSION_STAGE = "AWSCURRENT";
9    private final AWSSecretsManager secretsManager;
10
11    @Autowired
12    public AWSCredentialService(final AWSSecretsManager secretsManager) {
13        this.secretsManager = secretsManager;
14    }
15
16    @Override
17    public Optional<String> getSecret(@NotNull final String key) {
18        try {
19            final GetSecretValueResult secretValue = secretsManager.getSecretValue(new GetSecretValueRequest().withSecretId(key).withVersionStage(VERSION_STAGE));
20            return Optional.ofNullable(secretValue.getSecretString());
21        } catch (final ResourceNotFoundException e) {
22            log.debug("Unable to fetch secret [{}]: [{}]", key, e.getMessage());
23        }
24        return Optional.empty();
25    }
26}

1fluent-connect:
2  credential:
3    credential-manager: "aws-secrets-manager"

1  job-scheduler:
2    job-scheduler-type: "new-scheduler"

1@Component
2@Slf4j
3@FulfilmentOptionTransformationStep(id = "pickup-location-enrichment", priority = 1)
4public class PickupLocationTransformation implements FulfilmentOptionTransformationChain<FulfilmentRequest, FulfilmentResponse> {
5
6    @Override
7    public FulfilmentResponse transform(final FulfilmentRequest request, final FulfilmentResponse response, final FluentContext context) {
8        //your enrichment logic here
9    }
10}

1  configuration:
2    settingsManager: "custom-configuration-service"

1@Slf4j
2@Component
3@HandlerInfo(
4    name = "retryExampleHandler",
5    route = "retryExample",
6    description = "Retry example",
7    props = {@HandlerProp(key = "some-key",value = "a-value")}
8)
9public class RetryExampleHandler implements MessageHandler {
10    @Override
11    public void processMessage(@NotNull final MessageHandlerContext context) {
12        try {
13          service.doSomeLogic()
14        } catch (SomeCustomException ex) {
15          throw new HandlerRetryException(e);  //Let the message be retried when getting SomeCustomException
16        }
17        //If a handler throws any other kind of exeception, the message will be move to DLQ
18
19    }
20}

1@Slf4j
2@Component
3@HandlerInfo(
4    name = "defaultNotificationHandler",
5    route = "notification",
6    description = "Handles notifications and publishes to their designated route",
7    props = {@HandlerProp(key = "some-key",value = "a-value")}
8)
9public class DefaultNotificationHandler implements NotificationHandler {}

1@Slf4j
2@Component
3@HandlerInfo(
4    name = "customNotificationHandler",
5    route = "notification",
6    priority = 100,
7    description = "A custom handler for notifications"
8)
9public class CustomNotificationHandler implements NotificationHandler {}

1@Slf4j
2@Component
3@HandlerInfo(
4    name = "defaultNotificationHandler",
5    route = "notification",
6    priority = 100,
7    description = "A custom handler for notifications"
8)
9public class CustomNotificationHandler implements NotificationHandler {}

1  routes:
2    notification:
3      # Scenario 1: Adding a new configuration to add a new route and re-use an existing handler. Props is not specified and defaults are used
4      - route: : "custom-notification"
5        handler-name: "defaultNotificationHandler"
6
7      # Scenario 2: Modifying the configuration of an existing route to override the properties
8      - route: : "notification"
9        handler-name: "defaultNotificationHandler"
10        props:
11          some-key: "some-other-value"
12          another-key: "yet-another-value"
13
14      # Scenario 3: Modify an existing route to use a different existing handler
15      - route: : "notification"
16        handler-name: "anotherNotificationHandler"
17        props:
18          some-key: "some-other-value"
19          another-key: "yet-another-value"