Issue 29094 listening workflows

dotCMS/src/main/java/com/dotcms/api/system/event/SystemEventType.java

@@ -262,5 +262,8 @@ public enum SystemEventType {
 	SESSION_LOGOUT,
 
 	// Analytics App
-	ANALYTICS_APP
+	ANALYTICS_APP,
+
+	// Progress of action tasks
+	PROGRESS
 }

dotCMS/src/main/java/com/dotcms/concurrent/DotConcurrentFactory.java

@@ -5,8 +5,12 @@
 import com.dotcms.concurrent.lock.DotKeyLockManagerBuilder;
 import com.dotcms.concurrent.lock.DotKeyLockManagerFactory;
 import com.dotcms.concurrent.lock.IdentifierStripedLock;
+import com.dotcms.concurrent.monitor.LoggingTaskMonitor;
+import com.dotcms.concurrent.monitor.TaskMonitor;
+import com.dotcms.concurrent.monitor.ToastTaskMonitor;
 import com.dotcms.util.ConversionUtils;
 import com.dotcms.util.ReflectionUtils;
+import com.dotmarketing.business.APILocator;
 import com.dotmarketing.exception.DotRuntimeException;
 import com.dotmarketing.init.DotInitScheduler;
 import com.dotmarketing.util.Config;
@@ -92,6 +96,8 @@ public class DotConcurrentFactory implements DotConcurrentFactoryMBean, Serializ
 
     public static final String LOCK_MANAGER = "IdentifierStripedLock";
 
+    private final Map<String, TaskMonitor> taskMonitorMap = new ConcurrentHashMap<>();
+
     /**
      * Used to keep the instance of the submitter
      * Should be volatile to avoid thread-caching
@@ -191,6 +197,9 @@ private DotConcurrentFactory () {
         this.delayQueueConsumer = new DelayQueueConsumer();
         getScheduledThreadPoolExecutor().scheduleWithFixedDelay(this.delayQueueConsumer, 0,
                 Config.getLongProperty("dotcms.concurrent.delayqueue.waitmillis", 100), TimeUnit.MILLISECONDS);
+        this.taskMonitorMap.put("logging", new LoggingTaskMonitor());
+        this.taskMonitorMap.put("toast",   new LoggingTaskMonitor());
+        this.taskMonitorMap.put("progress", new LoggingTaskMonitor());
     }
 
     private void subscribeDelayQueue(final BlockingQueue<DelayedDelegate> delayedQueue) {
@@ -315,6 +324,15 @@ public static <T>T get (final Future<T> future) {
         }
     }
 
+    /**
+     * Return the current task id based on the server id and the thread id
+     * @return Object
+     */
+    public Object getCurrentTaskId() {
+
+        return APILocator.getServerAPI().readServerId() + Thread.currentThread().getId();
+    }
+
     private static class SingletonHolder {
         private static final DotConcurrentFactory INSTANCE = new DotConcurrentFactory();
     }
@@ -528,6 +546,16 @@ public IdentifierStripedLock getIdentifierStripedLock(){
        return this.identifierStripedLock;
     }
 
+/**
+     * Get the task monitor
+     * @param taskMonitorName {@link String}
+     * @return TaskMonitor
+     */
+    public TaskMonitor getTaskMonitor(final String taskMonitorName) {
+
+        final String defaultTaskMonitorName = Config.getStringProperty("DOT_DEFAULT_TASK_MONITOR_NAME","logging");
+        return this.taskMonitorMap.getOrDefault(taskMonitorName, this.taskMonitorMap.get(defaultTaskMonitorName));
+    }
     /**
      * Create a composite completable futures and results any of the first results done of the futures parameter.
      * @param futures

dotCMS/src/main/java/com/dotcms/concurrent/monitor/LoggingTaskMonitor.java

@@ -0,0 +1,34 @@
+package com.dotcms.concurrent.monitor;
+
+import com.dotmarketing.util.Logger;
+
+/**
+ * This is a simple implementation of the TaskMonitor interface that does just logging of the progress.
+ * @author jsanca
+ */
+public class LoggingTaskMonitor implements TaskMonitor {
+
+    @Override
+    public void onTaskStarted(final Object processId, final Object subProcessId) {
+
+        Logger.debug(this.getClass(), "Task started, process id: " + processId + " - sub process id:" + subProcessId);
+    }
+
+    @Override
+    public void onTaskProgress(final Object processId, final Object subProcessId, final int progress) {
+
+        Logger.debug(this.getClass(), "Task progress, process id: " + processId + " - sub process id:" + subProcessId + " - " + progress + "%");
+    }
+
+    @Override
+    public void onTaskCompleted(final Object processId, final Object subProcessId) {
+
+        Logger.debug(this.getClass(), "Task completed, process id: " + processId + " - sub process id:" + subProcessId);
+    }
+
+    @Override
+    public void onTaskFailed(final Object processId, final Object subProcessId, final Exception error) {
+
+        Logger.error(this.getClass(), "Task failed, process id: " + processId  + " - sub process id:" + subProcessId, error);
+    }
+}

dotCMS/src/main/java/com/dotcms/concurrent/monitor/ProgressBean.java

@@ -0,0 +1,101 @@
+package com.dotcms.concurrent.monitor;
+
+import java.io.Serializable;
+
+/**
+ * Class to manifest the progress of a bean
+ * @author jsanca
+ */
+public class ProgressBean implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+    private final int progress;
+    private final String message;
+    private final boolean completed;
+    private final boolean failed;
+    private final Object processId;
+    private final Object subProcessId;
+
+    public ProgressBean(final Builder builder) {
+        super();
+        this.progress = builder.progress;
+        this.message = builder.message;
+        this.completed = builder.completed;
+        this.failed = builder.failed;
+        this.processId = builder.processId;
+        this.subProcessId = builder.subProcessId;
+    }
+
+    public int getProgress() {
+        return progress;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public boolean isCompleted() {
+        return completed;
+    }
+
+    public boolean isFailed() {
+        return failed;
+    }
+
+    public Object getProcessId() {
+        return processId;
+    }
+
+    public Object getSubProcessId() {
+        return subProcessId;
+    }
+
+    public static class Builder {
+
+        private int progress;
+        private String message;
+        private boolean completed;
+        private boolean failed;
+        private Object processId;
+        private Object subProcessId;
+
+        public Builder progress(final int progress) {
+            this.progress = progress;
+            return this;
+        }
+
+        public Builder message(final String message) {
+            this.message = message;
+            return this;
+        }
+
+        public Builder completed(final boolean completed) {
+            this.completed = completed;
+            return this;
+        }
+
+        public Builder failed(final boolean failed) {
+            this.failed = failed;
+            return this;
+        }
+
+        public Builder processId(final Object processId) {
+            this.processId = processId;
+            return this;
+        }
+
+        public Builder subProcessId(final Object subProcessId) {
+            this.subProcessId = subProcessId;
+            return this;
+        }
+
+        public ProgressBean build() {
+            return new ProgressBean(this);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "ProgressBean [progress=" + progress + ", message=" + message + ", completed=" + completed + ", failed=" + failed + ", processId=" + processId + ", subProcessId=" + subProcessId + "]";
+    }
+}

dotCMS/src/main/java/com/dotcms/concurrent/monitor/ProgressTaskMonitor.java

@@ -0,0 +1,70 @@
+package com.dotcms.concurrent.monitor;
+
+import com.dotcms.api.system.event.Payload;
+import com.dotcms.api.system.event.SystemEvent;
+import com.dotcms.api.system.event.SystemEventType;
+import com.dotcms.api.system.event.SystemEventsAPI;
+import com.dotcms.api.system.event.Visibility;
+import com.dotcms.api.system.event.message.MessageSeverity;
+import com.dotcms.api.system.event.message.builder.SystemMessage;
+import com.dotcms.api.system.event.message.builder.SystemMessageBuilder;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.util.Logger;
+
+/**
+ * This is an implementation of the TaskMonitor interface that sends to the client
+ * Progress Beans undert the Progress system event type
+ * @author jsanca
+ */
+public class ProgressTaskMonitor implements TaskMonitor {
+
+    private final SystemEventsAPI systemEventsAPI = APILocator.getSystemEventsAPI();
+
+    @Override
+    public void onTaskStarted(final Object processId, final Object subProcessId) {
+
+        this.push(new ProgressBean.Builder()
+                .processId(processId).
+                subProcessId(subProcessId).build());
+    }
+
+    private void push (final ProgressBean progressBean) {
+
+        try {
+
+            this.systemEventsAPI.push(new SystemEvent(SystemEventType.PROGRESS,
+                    new Payload(progressBean)));
+        } catch (DotDataException e) {
+            Logger.error(this.getClass(), "Error pushing system event", e);
+        }
+    }
+
+    @Override
+    public void onTaskProgress(final Object processId, final Object subProcessId, final int progress) {
+
+        this.push(new ProgressBean.Builder()
+                .processId(processId).
+                subProcessId(subProcessId)
+                .progress(progress).build());
+    }
+
+    @Override
+    public void onTaskCompleted(final Object processId, final Object subProcessId) {
+
+        this.push(new ProgressBean.Builder()
+                .processId(processId).
+                subProcessId(subProcessId)
+                .completed(true).build());
+    }
+
+    @Override
+    public void onTaskFailed(final Object processId, final Object subProcessId, final Exception error) {
+
+        this.push(new ProgressBean.Builder()
+                .processId(processId).
+                subProcessId(subProcessId)
+                .failed(true)
+                .message(error.getMessage()).build());
+    }
+}

dotCMS/src/main/java/com/dotcms/concurrent/monitor/TaskMonitor.java

@@ -0,0 +1,33 @@
+package com.dotcms.concurrent.monitor;
+
+/**
+ * Interface for monitoring the progress of a task/process.
+
+ * @author jsanca
+ */
+public interface TaskMonitor {
+
+    /**
+     * Called when the task starts.
+     */
+    void onTaskStarted(Object processId, Object subProcessId);
+
+    /**
+     * Called to update the progress of the task.
+     *
+     * @param progress The current progress of the task, represented as a percentage (0-100).
+     */
+    void onTaskProgress(Object processId, Object subProcessId, int progress);
+
+    /**
+     * Called when the task completes successfully.
+     */
+    void onTaskCompleted(Object processId, Object subProcessId);
+
+    /**
+     * Called when the task fails.
+     *
+     * @param error The error that caused the task to fail.
+     */
+    void onTaskFailed(Object processId, Object subProcessId, Exception error);
+}

dotCMS/src/main/java/com/dotcms/concurrent/monitor/ToastTaskMonitor.java

@@ -0,0 +1,61 @@
+package com.dotcms.concurrent.monitor;
+
+import com.dotcms.api.system.event.Payload;
+import com.dotcms.api.system.event.SystemEvent;
+import com.dotcms.api.system.event.SystemEventType;
+import com.dotcms.api.system.event.SystemEventsAPI;
+import com.dotcms.api.system.event.Visibility;
+import com.dotcms.api.system.event.message.MessageSeverity;
+import com.dotcms.api.system.event.message.builder.SystemMessage;
+import com.dotcms.api.system.event.message.builder.SystemMessageBuilder;
+import com.dotmarketing.business.APILocator;
+import com.dotmarketing.exception.DotDataException;
+import com.dotmarketing.util.Logger;
+
+/**
+ * This is a simple implementation of the TaskMonitor interface that does toast messages of the progress.
+ * @author jsanca
+ */
+public class ToastTaskMonitor implements TaskMonitor {
+
+    private final SystemEventsAPI systemEventsAPI = APILocator.getSystemEventsAPI();
+
+    @Override
+    public void onTaskStarted(final Object processId, final Object subProcessId) {
+
+        this.sendMessage("Task started, process id: " + processId + " - sub process id:" + subProcessId);
+    }
+
+    private void sendMessage (final String message) {
+
+        try {
+
+            final SystemMessage systemMessage   = new SystemMessageBuilder()
+                    .setMessage(message)
+                    .setLife(15000) // 15 secs for testing
+                    .setSeverity(MessageSeverity.INFO).create();
+            this.systemEventsAPI.push(new SystemEvent(SystemEventType.MESSAGE,
+                    new Payload(systemMessage)));
+        } catch (DotDataException e) {
+            Logger.error(this.getClass(), "Error pushing system event", e);
+        }
+    }
+
+    @Override
+    public void onTaskProgress(final Object processId, final Object subProcessId, final int progress) {
+
+        this.sendMessage("Task progress, process id: " + processId + " - sub process id:" + subProcessId + " - " + progress + "%");
+    }
+
+    @Override
+    public void onTaskCompleted(final Object processId, final Object subProcessId) {
+
+        this.sendMessage("Task completed, process id: " + processId + " - sub process id:" + subProcessId);
+    }
+
+    @Override
+    public void onTaskFailed(final Object processId, final Object subProcessId, final Exception error) {
+
+        this.sendMessage("Task failed, process id: " + processId  + " - sub process id:" + subProcessId + " - " + error.getMessage());
+    }
+}