set default queue priority and range

Author: Yang Guo

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

@@ -21,6 +21,7 @@
 import com.flow.platform.core.queue.PriorityMessage;
 import com.flow.platform.core.queue.RabbitQueue;
 import com.flow.platform.util.Logger;
+import com.google.common.collect.Range;
 import javax.annotation.PostConstruct;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -36,9 +37,22 @@
 @Configuration
 public class QueueConfig {
 
-    public final static int CMD_QUEUE_MAX_LENGTH = 100;
+    public final static int QUEUE_MAX_LENGTH = 100;
 
-    public final static int CMD_QUEUE_DEFAULT_PRIORITY = 10;
+    /**
+     * Default queue priority
+     */
+    public final static int DEFAULT_PRIORITY = 1;
+
+    /**
+     * The max queue priority for special case like retry
+     */
+    public final static int MAX_PRIORITY = 100;
+
+    /**
+     * The priority range for queue
+     */
+    public final static Range PRIORITY_RANGE = Range.closed(1, 10);
 
     public final static String PROP_CMD_QUEUE_RETRY = "queue.cmd.retry.enable";
 
@@ -96,18 +110,18 @@ public void init() {
     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);
+            return new RabbitQueue(taskExecutor, host, QUEUE_MAX_LENGTH, DEFAULT_PRIORITY, cmdQueueName);
         }
 
         LOGGER.trace("Apply in memory queue for cmd queue");
-        return new InMemoryQueue<>(taskExecutor, CMD_QUEUE_MAX_LENGTH, "CmdQueue");
+        return new InMemoryQueue<>(taskExecutor, QUEUE_MAX_LENGTH, "CmdQueue");
     }
 
     /**
      * Queue to handle cmd status update
      */
     @Bean
     public PlatformQueue<PriorityMessage> cmdStatusQueue() {
-        return new InMemoryQueue<>(taskExecutor, 100, "CmdStatusQueue");
+        return new InMemoryQueue<>(taskExecutor, QUEUE_MAX_LENGTH, "CmdStatusQueue");
     }
 }

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

@@ -161,7 +161,7 @@ private void resend(final CmdQueueItem item, final int retry) {
         }
 
         // reset cmd status
-        PriorityMessage message = PriorityMessage.create(item.toBytes(), QueueConfig.CMD_QUEUE_DEFAULT_PRIORITY);
+        PriorityMessage message = PriorityMessage.create(item.toBytes(), QueueConfig.MAX_PRIORITY);
         cmdQueue.enqueue(message);
         LOGGER.trace("Re-enqueue item %s", item);
     }

platform-control-center/src/main/java/com/flow/platform/cc/controller/CmdController.java

@@ -16,6 +16,7 @@
 
 package com.flow.platform.cc.controller;
 
+import com.flow.platform.cc.config.QueueConfig;
 import com.flow.platform.cc.domain.CmdStatusItem;
 import com.flow.platform.cc.service.CmdDispatchService;
 import com.flow.platform.cc.service.CmdService;
@@ -73,7 +74,7 @@ public Cmd sendCommand(@RequestBody CmdInfo cmd) {
 
     @PostMapping(path = "/queue/send")
     public Cmd sendCommandToQueue(@RequestBody CmdInfo cmd, @RequestParam int priority, @RequestParam int retry) {
-        if (!Range.closed(1, 10).contains(priority)) {
+        if (!QueueConfig.PRIORITY_RANGE.contains(priority)) {
             throw new IllegalParameterException("Illegal priority value should between (1 - 10)");
         }
 

platform-control-center/src/main/java/com/flow/platform/cc/service/CmdServiceImpl.java

@@ -24,6 +24,7 @@
 import static com.flow.platform.domain.CmdType.STOP;
 
 import com.flow.platform.cc.config.AppConfig;
+import com.flow.platform.cc.config.QueueConfig;
 import com.flow.platform.cc.dao.AgentDao;
 import com.flow.platform.cc.dao.CmdDao;
 import com.flow.platform.cc.dao.CmdLogDao;
@@ -216,7 +217,7 @@ public Cmd enqueue(CmdInfo cmdInfo, int priority, int retry) {
     public void updateStatus(CmdStatusItem statusItem, boolean inQueue) {
         if (inQueue) {
             LOGGER.trace("Report cmd status from queue: %s", statusItem.getCmdId());
-            cmdStatusQueue.enqueue(PriorityMessage.create(statusItem.toBytes(), 1));
+            cmdStatusQueue.enqueue(PriorityMessage.create(statusItem.toBytes(), QueueConfig.DEFAULT_PRIORITY));
             return;
         }