refactor on priority queue item with timestamp

Author: Yang Guo

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

@@ -31,6 +31,8 @@
 @Configurable
 public class QueueConfig {
 
+    public final static long DEFAULT_CMD_CALLBACK_QUEUE_PRIORITY = 1L;
+
     @Autowired
     private ThreadPoolTaskExecutor taskExecutor;
 

platform-api/src/main/java/com/flow/platform/api/controller/CmdWebhookController.java

@@ -16,6 +16,7 @@
 
 package com.flow.platform.api.controller;
 
+import com.flow.platform.api.config.QueueConfig;
 import com.flow.platform.api.domain.CmdCallbackQueueItem;
 import com.flow.platform.api.service.job.JobService;
 import com.flow.platform.core.exception.IllegalParameterException;
@@ -64,7 +65,7 @@ public void execute(@RequestBody Cmd cmd, @RequestParam String identifier) {
                 cmd.getStatus(),
                 cmd.getId(),
                 cmd.getCmdResult());
-            jobService.enqueue(new CmdCallbackQueueItem(jobId, cmd), System.nanoTime());
+            jobService.enqueue(new CmdCallbackQueueItem(jobId, cmd), QueueConfig.DEFAULT_CMD_CALLBACK_QUEUE_PRIORITY);
         } catch (NumberFormatException warn) {
             LOGGER.warn("Invalid job id format");
         }

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

@@ -98,27 +98,22 @@ public void onQueueItem(PriorityMessage message) {
             cmd.setRetry(retry);
             cmdService.save(cmd);
 
-            // re-enqueue
-            resend(cmd.getId(), retry);
+            // do not retry
+            if (retry <= 0) {
+                return;
+            }
+
+            // retry the message
+            retry(message);
 
         } catch (Throwable e) {
             LOGGER.error("Unexpected exception", e);
         }
     }
 
-    /**
-     * Re-enqueue cmd and return num of retry
-     */
-    private void resend(final String cmdId, final int retry) {
-        if (retry <= 0) {
-            return;
-        }
-
-        ThreadUtil.sleep(RETRY_WAIT_TIME);
-
-        // reset cmd status
-        PriorityMessage message = PriorityMessage.create(cmdId.getBytes(), QueueConfig.MAX_PRIORITY);
+    private void retry(final PriorityMessage message) {
+        message.setPriority(QueueConfig.MAX_PRIORITY);
         cmdQueue.enqueue(message);
-        LOGGER.trace("Re-enqueue item %s", cmdId);
+        ThreadUtil.sleep(RETRY_WAIT_TIME);
     }
 }

platform-core/src/main/java/com/flow/platform/core/queue/PriorityMessage.java

@@ -27,27 +27,44 @@ public class PriorityMessage extends Message implements PriorityQueueItem {
 
     private Long priority;
 
+    private Long timestamp;
+
     public static PriorityMessage create(byte[] content, long priority) {
         MessageProperties properties = new MessageProperties();
         return new PriorityMessage(content, properties, priority);
     }
 
     public PriorityMessage(Message message) {
         super(message.getBody(), message.getMessageProperties());
+        this.timestamp = System.nanoTime();
     }
 
     public PriorityMessage(byte[] body, MessageProperties messageProperties, long priority) {
         super(body, messageProperties);
         this.priority = priority;
+        this.timestamp = System.currentTimeMillis();
     }
 
     @Override
     public Long getPriority() {
         return this.priority;
     }
 
+    public void setPriority(long priority) {
+        this.priority = priority;
+    }
+
+    @Override
+    public Long getTimestamp() {
+        return this.timestamp;
+    }
+
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
     @Override
     public int compareTo(PriorityQueueItem o) {
-        return o.getPriority().compareTo(getPriority());
+        return COMPARATOR.compare(this, o);
     }
 }

platform-core/src/test/java/com/flow/platform/core/test/PlatformQueueTest.java

@@ -17,6 +17,7 @@
 package com.flow.platform.core.test;
 
 import com.flow.platform.core.queue.PriorityMessage;
+import com.flow.platform.core.util.ThreadUtil;
 import com.flow.platform.queue.PlatformQueue;
 import com.flow.platform.queue.QueueListener;
 import com.flow.platform.util.ObjectWrapper;
@@ -87,7 +88,7 @@ public void should_enqueue_for_in_memory_queue() throws Throwable {
     @Test
     public void should_enqueue_with_priority_in_memory_queue() throws Throwable {
         // given: queue listener
-        final int size = 2;
+        final int size = 4;
         CountDownLatch latch = new CountDownLatch(size);
         List<String> prioritizedList = new ArrayList<>(size);
 
@@ -98,17 +99,24 @@ public void should_enqueue_with_priority_in_memory_queue() throws Throwable {
 
         // when:
         inMemoryQueue.register(listener);
-        inMemoryQueue.enqueue(PriorityMessage.create("1".getBytes(), 1));  // should be second
-        inMemoryQueue.enqueue(PriorityMessage.create("2".getBytes(), 10)); // should be first
+        inMemoryQueue.enqueue(PriorityMessage.create("1".getBytes(), 1));
+        ThreadUtil.sleep(1);
+        inMemoryQueue.enqueue(PriorityMessage.create("2".getBytes(), 1));
+        ThreadUtil.sleep(1);
+        inMemoryQueue.enqueue(PriorityMessage.create("3".getBytes(), 10));
+        ThreadUtil.sleep(1);
+        inMemoryQueue.enqueue(PriorityMessage.create("4".getBytes(), 10));
         inMemoryQueue.start();
 
         // then:
         boolean await = latch.await(60, TimeUnit.SECONDS);
         Assert.assertTrue(await);
         Assert.assertEquals(size, prioritizedList.size());
 
-        Assert.assertEquals("2", prioritizedList.get(0));
-        Assert.assertEquals("1", prioritizedList.get(1));
+        Assert.assertEquals("4", prioritizedList.get(0));
+        Assert.assertEquals("3", prioritizedList.get(1));
+        Assert.assertEquals("2", prioritizedList.get(2));
+        Assert.assertEquals("1", prioritizedList.get(3));
     }
 
     @Test

platform-queue/src/main/java/com/flow/platform/queue/DefaultQueueMessage.java

@@ -25,23 +25,31 @@ public class DefaultQueueMessage implements PriorityQueueItem {
 
     private Long priority;
 
+    private Long timestamp;
+
     public DefaultQueueMessage(byte[] body, Long priority) {
         this.body = body;
         this.priority = priority;
+        this.timestamp = System.currentTimeMillis();
     }
 
     @Override
     public Long getPriority() {
         return priority;
     }
 
+    @Override
+    public Long getTimestamp() {
+        return timestamp;
+    }
+
     @Override
     public byte[] getBody() {
-        return new byte[0];
+        return body;
     }
 
     @Override
     public int compareTo(PriorityQueueItem o) {
-        return o.getPriority().compareTo(getPriority());
+        return COMPARATOR.compare(this, o);
     }
 }

platform-queue/src/main/java/com/flow/platform/queue/PriorityQueueItem.java

@@ -16,12 +16,31 @@
 
 package com.flow.platform.queue;
 
+import java.util.Comparator;
+import java.util.Objects;
+
 /**
  * @author yang
  */
 public interface PriorityQueueItem extends Comparable<PriorityQueueItem> {
 
+    ItemComparator COMPARATOR = new ItemComparator();
+
     Long getPriority();
 
+    Long getTimestamp();
+
     byte[] getBody();
+
+    class ItemComparator implements Comparator<PriorityQueueItem> {
+
+        @Override
+        public int compare(PriorityQueueItem o1, PriorityQueueItem o2) {
+            if (Objects.equals(o1.getPriority(), o2.getPriority())) {
+                return o2.getTimestamp().compareTo(o1.getTimestamp());
+            }
+
+            return o2.getPriority().compareTo(o1.getPriority());
+        }
+    }
 }