Refactor connector context with strategy pattern

Refactored DumConnectorContextImpl to use a strategy and chain-of-responsibility pattern for job item processing. Introduced JobProcessingChain and JobProcessingStrategy interfaces with implementations for index, reindex, deindex, ignore rules, and unchanged strategies. Added JobItemBatchProcessor for efficient batch queueing. Updated related classes and dependency injection. Updated POM files to version 2026.1.7 and added spring-boot-starter-json dependency.

Author: alegauss

aem-commons/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.viglet.dumont</groupId>
         <artifactId>dumont</artifactId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <groupId>com.viglet.dumont.connector</groupId>

aem/aem-plugin-sample/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.viglet.dumont.connector</groupId>
         <artifactId>dumont-aem</artifactId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <artifactId>aem-plugin-sample</artifactId>

aem/aem-plugin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.viglet.dumont.connector</groupId>
         <artifactId>dumont-aem</artifactId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <artifactId>aem-plugin</artifactId>

aem/aem-plugin/src/main/java/com/viglet/dumont/connector/plugin/aem/DumAemPluginProcess.java

@@ -115,7 +115,7 @@ public void sentToIndexStandalone(@NotNull String source,
                                 dumAemPathList.getPaths());
                 dumAemSourceService.getDumAemSourceByName(source).ifPresentOrElse(dumAemSource -> {
                         DumAemSession dumAemSession = dumAemSessionService.getDumAemSession(dumAemSource,
-                                        dumAemPathList);
+                                        dumAemPathList, true);
                         indexContentIdList(dumAemPathList.getPaths(), dumAemSession);
                         if (connectorDependencies) {
                                 indexDependencies(dumAemSession, dumAemPathList.getPaths());
@@ -187,7 +187,7 @@ public void indexAll(DumAemSource dumAemSource) {
                         return;
                 }
                 runningSources.add(dumAemSource.getName());
-                DumAemSession dumAemSession = dumAemSessionService.getDumAemSession(dumAemSource);
+                DumAemSession dumAemSession = dumAemSessionService.getDumAemSession(dumAemSource, false);
                 try {
                         this.getNodesFromJson(dumAemSession);
                 } catch (Exception e) {
@@ -204,8 +204,13 @@ private void indexDependencies(DumAemSession dumAemSession, List<String> idList)
         }
 
         private void finished(DumAemSession dumAemSession) {
-                if (!dumAemSession.isStandalone())
+                if (!dumAemSession.isStandalone()) {
+                        log.info("Stopping source process. {}", dumAemSession.getSource());
                         runningSources.remove(dumAemSession.getSource());
+                } else {
+                        log.info("Finished standalone indexing for source: {}",
+                                        dumAemSession.getSource());
+                }
                 dumConnectorContext.finishIndexing(dumAemSession, dumAemSession.isStandalone());
         }
 

aem/aem-plugin/src/main/java/com/viglet/dumont/connector/plugin/aem/service/DumAemSessionService.java

@@ -30,7 +30,7 @@ public DumAemSessionService(DumAemSourceService dumAemSourceService,
         }
 
         public DumAemSession getDumAemSession(DumAemSource dumAemSource,
-                        DumAemPathList dumAemPathList) {
+                        DumAemPathList dumAemPathList, boolean standalone) {
                 // Retrieve content mapping once and reuse
                 DumAemContentMapping dumAemContentMapping = dumAemContentMappingService
                                 .getDumAemContentMapping(dumAemSource);
@@ -60,7 +60,7 @@ public DumAemSession getDumAemSession(DumAemSource dumAemSource,
                 return DumAemSession.builder()
                                 .configuration(dumAemConfiguration)
                                 .event(event)
-                                .standalone(true)
+                                .standalone(standalone)
                                 .indexChildren(indexChildren)
                                 .source(session.getSource())
                                 .transactionId(session.getTransactionId())
@@ -73,8 +73,7 @@ public DumAemSession getDumAemSession(DumAemSource dumAemSource,
                                 .build();
         }
 
-        public DumAemSession getDumAemSession(DumAemSource dumAemSource) {
-                return getDumAemSession(dumAemSource, (DumAemPathList) null);
-
+        public DumAemSession getDumAemSession(DumAemSource dumAemSource, boolean standalone) {
+                return getDumAemSession(dumAemSource, (DumAemPathList) null, standalone);
         }
 }

aem/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.viglet.dumont</groupId>
         <artifactId>dumont</artifactId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <groupId>com.viglet.dumont.connector</groupId>

commons/pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>com.viglet.dumont</groupId>
         <artifactId>dumont</artifactId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <artifactId>dumont-commons</artifactId>

connector/connector-app/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>dumont-connector</artifactId>
         <groupId>com.viglet.dumont.connector</groupId>
-        <version>2026.1.4</version>
+        <version>2026.1.7</version>
     </parent>
 
     <artifactId>connector-app</artifactId>
@@ -52,12 +52,6 @@
         <dependency>
             <groupId>com.viglet.turing</groupId>
             <artifactId>turing-java-sdk</artifactId>
-            <exclusions>
-                <exclusion>
-                    <groupId>com.fasterxml.jackson.datatype</groupId>
-                    <artifactId>jackson-datatype-jdk8</artifactId>
-                </exclusion>
-            </exclusions>
         </dependency>
         <dependency>
             <groupId>com.sezinkarli</groupId>
@@ -102,6 +96,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-hazelcast</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>

connector/connector-app/src/main/java/com/viglet/dumont/connector/batch/JobItemBatchProcessor.java

@@ -0,0 +1,130 @@
+/*
+ *
+ * Copyright (C) 2016-2024 the original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
+ * even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with this program. If
+ * not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.viglet.dumont.connector.batch;
+
+import static com.viglet.dumont.commons.indexing.DumIndexingStatus.SENT_TO_QUEUE;
+import static com.viglet.dumont.commons.sn.field.DumSNFieldName.ID;
+import static com.viglet.dumont.connector.commons.logging.DumConnectorLoggingUtils.setSuccessStatus;
+import static com.viglet.dumont.connector.constant.DumConnectorConstants.CONNECTOR_INDEXING_QUEUE;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.jms.core.JmsMessagingTemplate;
+import org.springframework.stereotype.Component;
+
+import com.google.common.collect.Iterators;
+import com.viglet.dumont.connector.commons.DumConnectorSession;
+import com.viglet.turing.client.sn.job.TurSNJobItem;
+import com.viglet.turing.client.sn.job.TurSNJobItems;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Batch processor for managing job items efficiently.
+ * Handles buffering and sending job items to the message queue.
+ * 
+ * @author Alexandre Oliveira
+ * @since 2026.1
+ */
+@Slf4j
+@Component
+public class JobItemBatchProcessor {
+
+    private final TurSNJobItems buffer = new TurSNJobItems();
+    private final JmsMessagingTemplate jmsMessagingTemplate;
+    private final int batchSize;
+
+    public JobItemBatchProcessor(
+            @Value("${dumont.job.size:50}") int batchSize,
+            JmsMessagingTemplate jmsMessagingTemplate) {
+        this.batchSize = batchSize;
+        this.jmsMessagingTemplate = jmsMessagingTemplate;
+    }
+
+    /**
+     * Adds a job item to the buffer.
+     * Automatically flushes when batch size is reached.
+     * 
+     * @param item    the job item to add
+     * @param session the connector session
+     */
+    public synchronized void add(TurSNJobItem item, DumConnectorSession session) {
+        buffer.add(item);
+        logQueueInfo();
+
+        if (buffer.size() >= batchSize) {
+            flush(session);
+        }
+    }
+
+    /**
+     * Flushes all buffered items to the message queue.
+     * 
+     * @param session the connector session
+     */
+    public synchronized void flush(DumConnectorSession session) {
+        if (buffer.size() == 0) {
+            log.info("No job to send to connector queue.");
+            return;
+        }
+
+        log.info("Sending {} jobs to connector queue.", buffer.size());
+
+        if (log.isDebugEnabled()) {
+            for (TurSNJobItem turSNJobItem : buffer) {
+                log.debug("TurSNJobItem Id: {}", turSNJobItem.getAttributes().get(ID));
+            }
+        }
+
+        // Update status for all items
+        for (TurSNJobItem turSNJobItem : buffer) {
+            setSuccessStatus(turSNJobItem, session, SENT_TO_QUEUE);
+        }
+
+        // Create a copy to send
+        TurSNJobItems itemsToSend = new TurSNJobItems();
+        for (TurSNJobItem turSNJobItem : buffer) {
+            itemsToSend.add(turSNJobItem);
+        }
+
+        jmsMessagingTemplate.convertAndSend(CONNECTOR_INDEXING_QUEUE, itemsToSend);
+        buffer.clear();
+
+        log.info("Successfully sent batch to connector queue.");
+    }
+
+    /**
+     * Returns the current buffer size.
+     * 
+     * @return the number of items in the buffer
+     */
+    public synchronized int size() {
+        return buffer.size();
+    }
+
+    /**
+     * Checks if the buffer is empty.
+     * 
+     * @return true if buffer has no items
+     */
+    public synchronized boolean isEmpty() {
+        return buffer.size() == 0;
+    }
+
+    private void logQueueInfo() {
+        log.debug("Total Job Items in buffer: {}", Iterators.size(buffer.iterator()));
+    }
+}

connector/connector-app/src/main/java/com/viglet/dumont/connector/chain/JobProcessingChain.java

@@ -0,0 +1,91 @@
+/*
+ *
+ * Copyright (C) 2016-2024 the original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
+ * even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with this program. If
+ * not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.viglet.dumont.connector.chain;
+
+import java.util.Comparator;
+import java.util.List;
+
+import org.springframework.stereotype.Component;
+
+import com.viglet.dumont.connector.batch.JobItemBatchProcessor;
+import com.viglet.dumont.connector.commons.domain.DumJobItemWithSession;
+import com.viglet.dumont.connector.strategy.JobProcessingStrategy;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Chain of Responsibility implementation for processing job items.
+ * Routes job items to the appropriate strategy based on priority and
+ * capability.
+ * 
+ * @author Alexandre Oliveira
+ * @since 2026.1
+ */
+@Slf4j
+@Component
+public class JobProcessingChain {
+
+    private final List<JobProcessingStrategy> strategies;
+
+    public JobProcessingChain(List<JobProcessingStrategy> strategies) {
+        // Sort strategies by priority (lower values first)
+        this.strategies = strategies.stream()
+                .sorted(Comparator.comparingInt(JobProcessingStrategy::getPriority))
+                .toList();
+
+        log.info("Initialized JobProcessingChain with {} strategies", strategies.size());
+        strategies.forEach(strategy -> log.debug("Strategy: {} with priority {}",
+                strategy.getClass().getSimpleName(),
+                strategy.getPriority()));
+    }
+
+    /**
+     * Processes a job item by finding and executing the first matching strategy.
+     * 
+     * @param jobItem        the job item to process
+     * @param batchProcessor the batch processor for queueing items
+     */
+    public void process(DumJobItemWithSession jobItem, JobItemBatchProcessor batchProcessor) {
+        if (jobItem == null || jobItem.turSNJobItem() == null) {
+            log.warn("Received null job item, skipping processing");
+            return;
+        }
+
+        strategies.stream()
+                .filter(strategy -> strategy.canHandle(jobItem))
+                .findFirst()
+                .ifPresentOrElse(
+                        strategy -> {
+                            log.debug("Processing {} with {}",
+                                    jobItem.turSNJobItem().getId(),
+                                    strategy.getClass().getSimpleName());
+                            strategy.process(jobItem, batchProcessor);
+                        },
+                        () -> log.warn("No strategy found for job item: {} (action: {})",
+                                jobItem.turSNJobItem().getId(),
+                                jobItem.turSNJobItem().getTurSNJobAction()));
+    }
+
+    /**
+     * Returns the list of registered strategies.
+     * 
+     * @return list of strategies sorted by priority
+     */
+    public List<JobProcessingStrategy> getStrategies() {
+        return strategies;
+    }
+}