apply priority message for queue item

Author: Yang Guo

platform-agent/src/main/java/com/flow/platform/agent/CmdManager.java

@@ -258,6 +258,7 @@ public synchronized void kill() {
         for (Map.Entry<Cmd, CmdResult> entry : running.entrySet()) {
             CmdResult r = entry.getValue();
             Cmd cmd = entry.getKey();
+            finished.put(cmd, r);
 
             r.getProcess().destroy();
 

platform-api/src/main/java/com/flow/platform/api/config/AppConfig.java

@@ -23,6 +23,7 @@
 import com.flow.platform.core.config.DatabaseConfig;
 import com.flow.platform.core.queue.InMemoryQueue;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.util.Logger;
 import java.io.IOException;
@@ -34,15 +35,12 @@
 import org.apache.velocity.app.VelocityEngine;
 import org.apache.velocity.runtime.RuntimeConstants;
 import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.event.ApplicationEventMulticaster;
 import org.springframework.context.event.SimpleApplicationEventMulticaster;
-import org.springframework.scheduling.TaskScheduler;
-import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
@@ -136,7 +134,7 @@ public ThreadPoolTaskExecutor taskExecutor() {
      * Queue to process cmd callback task
      */
     @Bean
-    public PlatformQueue<CmdCallbackQueueItem> cmdCallbackQueue() {
+    public PlatformQueue<PriorityMessage> cmdCallbackQueue() {
         return new InMemoryQueue<>(executor, 50, "CmdCallbackQueue");
     }
 

platform-api/src/main/java/com/flow/platform/api/consumer/CmdCallbackQueueConsumer.java

@@ -20,6 +20,7 @@
 import com.flow.platform.api.service.job.JobService;
 import com.flow.platform.core.exception.NotFoundException;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.QueueListener;
 import com.flow.platform.util.Logger;
 import javax.annotation.PostConstruct;
@@ -30,12 +31,12 @@
  * @author yh@firim
  */
 @Component
-public class CmdCallbackQueueConsumer implements QueueListener<CmdCallbackQueueItem> {
+public class CmdCallbackQueueConsumer implements QueueListener<PriorityMessage> {
 
     private final static Logger LOGGER = new Logger(CmdCallbackQueueConsumer.class);
 
     @Autowired
-    private PlatformQueue<CmdCallbackQueueItem> cmdCallbackQueue;
+    private PlatformQueue<PriorityMessage> cmdCallbackQueue;
 
     @Autowired
     private JobService jobService;
@@ -46,10 +47,13 @@ public void init() {
     }
 
     @Override
-    public void onQueueItem(CmdCallbackQueueItem item) {
-        if (item == null) {
+    public void onQueueItem(PriorityMessage message) {
+        if (message == null) {
             return;
         }
+
+        CmdCallbackQueueItem item = CmdCallbackQueueItem.parse(message.getBody(), CmdCallbackQueueItem.class);
+
         try {
             jobService.callback(item);
         } catch (NotFoundException notFoundException) {

platform-api/src/main/java/com/flow/platform/api/domain/CmdCallbackQueueItem.java

@@ -17,12 +17,13 @@
 package com.flow.platform.api.domain;
 
 import com.flow.platform.domain.Cmd;
+import com.flow.platform.domain.Jsonable;
 import java.math.BigInteger;
 
 /**
  * @author yh@firim
  */
-public class CmdCallbackQueueItem {
+public class CmdCallbackQueueItem extends Jsonable {
 
     private final BigInteger jobId;
 

platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java

@@ -55,6 +55,7 @@
 import com.flow.platform.core.exception.IllegalStatusException;
 import com.flow.platform.core.exception.NotFoundException;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.service.ApplicationEventService;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdInfo;
@@ -108,7 +109,7 @@ public class JobServiceImpl extends ApplicationEventService implements JobServic
     private JobNodeService jobNodeService;
 
     @Autowired
-    private PlatformQueue<CmdCallbackQueueItem> cmdCallbackQueue;
+    private PlatformQueue<PriorityMessage> cmdCallbackQueue;
 
     @Autowired
     private JobDao jobDao;
@@ -458,7 +459,7 @@ private void onRunShellCallback(String path, Cmd cmd, Job job) {
 
     @Override
     public void enterQueue(CmdCallbackQueueItem cmdQueueItem) {
-        cmdCallbackQueue.enqueue(cmdQueueItem);
+        cmdCallbackQueue.enqueue(PriorityMessage.create(cmdQueueItem.toBytes(), 1));
     }
 
     @Override

platform-control-center/src/main/java/com/flow/platform/cc/config/QueueConfig.java

@@ -16,14 +16,12 @@
 
 package com.flow.platform.cc.config;
 
-import com.flow.platform.cc.domain.CmdStatusItem;
 import com.flow.platform.core.queue.InMemoryQueue;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.RabbitQueue;
-import com.flow.platform.domain.AgentPath;
 import com.flow.platform.util.Logger;
 import javax.annotation.PostConstruct;
-import org.springframework.amqp.core.Message;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -95,7 +93,7 @@ public void init() {
     }
 
     @Bean
-    public PlatformQueue<Message> cmdQueue() {
+    public PlatformQueue<PriorityMessage> cmdQueue() {
         if (cmdQueueRabbitEnable) {
             LOGGER.trace("Apply RabbitMQ for cmd queue");
             return new RabbitQueue(taskExecutor, host, CMD_QUEUE_MAX_LENGTH, CMD_QUEUE_DEFAULT_PRIORITY, cmdQueueName);
@@ -105,19 +103,11 @@ public PlatformQueue<Message> cmdQueue() {
         return new InMemoryQueue<>(taskExecutor, CMD_QUEUE_MAX_LENGTH, "CmdQueue");
     }
 
-    /**
-     * Queue to handle agent report online in sync
-     */
-    @Bean
-    public PlatformQueue<AgentPath> agentReportQueue() {
-        return new InMemoryQueue<>(taskExecutor, 100, "AgentReportQueue");
-    }
-
     /**
      * Queue to handle cmd status update
      */
     @Bean
-    public PlatformQueue<CmdStatusItem> cmdStatusQueue() {
+    public PlatformQueue<PriorityMessage> cmdStatusQueue() {
         return new InMemoryQueue<>(taskExecutor, 100, "CmdStatusQueue");
     }
 }

platform-control-center/src/main/java/com/flow/platform/cc/consumer/AgentReportQueueConsumer.java

@@ -21,6 +21,7 @@
 import com.flow.platform.cc.event.AgentResourceEvent.Category;
 import com.flow.platform.cc.service.ZoneService;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.util.Logger;
 import org.springframework.amqp.core.Message;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -40,7 +41,7 @@ public class AgentResourceEventHandler implements ApplicationListener<AgentResou
     private ZoneService zoneService;
 
     @Autowired
-    private PlatformQueue<Message> cmdQueue;
+    private PlatformQueue<PriorityMessage> cmdQueue;
 
     @Value("${queue.cmd.retry.enable}")
     private Boolean cmdQueueRetryEnable;

platform-control-center/src/main/java/com/flow/platform/cc/consumer/AgentResourceEventHandler.java

@@ -25,6 +25,7 @@
 import com.flow.platform.core.exception.IllegalParameterException;
 import com.flow.platform.core.exception.IllegalStatusException;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.QueueListener;
 import com.flow.platform.domain.Cmd;
 import com.flow.platform.domain.CmdStatus;
@@ -33,10 +34,6 @@
 import java.time.Duration;
 import java.time.Instant;
 import javax.annotation.PostConstruct;
-import org.springframework.amqp.core.Message;
-import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
@@ -47,7 +44,7 @@
  * @author [email protected]
  */
 @Component
-public class CmdQueueConsumer implements QueueListener<Message> {
+public class CmdQueueConsumer implements QueueListener<PriorityMessage> {
 
     private final static Logger LOGGER = new Logger(CmdQueueConsumer.class);
 
@@ -67,15 +64,15 @@ public class CmdQueueConsumer implements QueueListener<Message> {
     private AgentService agentService;
 
     @Autowired
-    private PlatformQueue<Message> cmdQueue;
+    private PlatformQueue<PriorityMessage> cmdQueue;
 
     @PostConstruct
     public void init() {
         cmdQueue.register(this);
     }
 
     @Override
-    public void onQueueItem(Message message) {
+    public void onQueueItem(PriorityMessage message) {
         CmdQueueItem item = CmdQueueItem.parse(message.getBody(), CmdQueueItem.class);
         String cmdId = item.getCmdId();
         LOGGER.trace("Receive a cmd queue item: %s", item);
@@ -155,7 +152,6 @@ private void resend(final CmdQueueItem item, final int retry) {
             return;
         }
 
-        item.setPriority(QueueConfig.CMD_QUEUE_DEFAULT_PRIORITY);
         item.setRetry(retry);
 
         try {
@@ -165,9 +161,7 @@ private void resend(final CmdQueueItem item, final int retry) {
         }
 
         // reset cmd status
-        MessageProperties properties = new MessageProperties();
-        properties.setPriority(item.getPriority());
-        Message message = new Message(item.toBytes(), properties);
+        PriorityMessage message = PriorityMessage.create(item.toBytes(), QueueConfig.CMD_QUEUE_DEFAULT_PRIORITY);
         cmdQueue.enqueue(message);
         LOGGER.trace("Re-enqueue item %s", item);
     }

platform-control-center/src/main/java/com/flow/platform/cc/consumer/CmdQueueConsumer.java

@@ -19,7 +19,9 @@
 import com.flow.platform.cc.domain.CmdStatusItem;
 import com.flow.platform.cc.service.CmdService;
 import com.flow.platform.core.queue.PlatformQueue;
+import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.QueueListener;
+import com.flow.platform.domain.CmdStatus;
 import com.flow.platform.util.Logger;
 import javax.annotation.PostConstruct;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -31,12 +33,12 @@
  * @author yang
  */
 @Component
-public class CmdStatusQueueConsumer implements QueueListener<CmdStatusItem> {
+public class CmdStatusQueueConsumer implements QueueListener<PriorityMessage> {
 
     private final static Logger LOGGER = new Logger(CmdStatusQueueConsumer.class);
 
     @Autowired
-    private PlatformQueue<CmdStatusItem> cmdStatusQueue;
+    private PlatformQueue<PriorityMessage> cmdStatusQueue;
 
     @Autowired
     private CmdService cmdService;
@@ -48,14 +50,15 @@ public void init() {
     }
 
     @Override
-    public void onQueueItem(CmdStatusItem item) {
+    public void onQueueItem(PriorityMessage item) {
         if (item == null) {
             return;
         }
 
         try {
+            CmdStatusItem statusItem = CmdStatusItem.parse(item.getBody(), CmdStatusItem.class);
             LOGGER.debug(Thread.currentThread().getName() + " : " + item.toString());
-            cmdService.updateStatus(item, false);
+            cmdService.updateStatus(statusItem, false);
         } catch (Throwable e) {
             LOGGER.error("Update cmd error:", e);
         }