add unit test for queue priority

Author: Yang Guo

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

@@ -91,6 +91,11 @@ public void resume() {
         pause = false;
     }
 
+    @Override
+    public void clean() {
+        queue.clear();
+    }
+
     @Override
     public boolean isRunning() {
         return !pause && !stop;

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

@@ -75,6 +75,11 @@ public void cleanListener() {
      */
     public abstract void resume();
 
+    /**
+     * Remove all items from queue
+     */
+    public abstract void clean();
+
     /**
      * Queue processor is running
      */

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

@@ -40,6 +40,6 @@ public PriorityMessage(byte[] body, MessageProperties messageProperties) {
 
     @Override
     public int compareTo(PriorityMessage o) {
-        return this.getMessageProperties().getPriority().compareTo(o.getMessageProperties().getPriority());
+        return o.getMessageProperties().getPriority().compareTo(this.getMessageProperties().getPriority());
     }
 }

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

@@ -99,6 +99,11 @@ public void enqueue(PriorityMessage item) {
         size.incrementAndGet();
     }
 
+    @Override
+    public void clean() {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean isRunning() {
         return container.isRunning();

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

@@ -19,14 +19,15 @@
 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.core.queue.RabbitQueue;
 import com.flow.platform.util.ObjectWrapper;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.springframework.amqp.core.Message;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
@@ -76,6 +77,31 @@ public void should_enqueue_for_in_memory_queue() throws Throwable {
         Assert.assertEquals(0, inMemoryQueue.size());
     }
 
+    @Test
+    public void should_enqueue_with_priority_in_memory_queue() throws Throwable {
+        // given: queue listener
+        int size = 2;
+        CountDownLatch latch = new CountDownLatch(size);
+        List<String> prioritizedList = new ArrayList<>(size);
+
+        QueueListener<PriorityMessage> listener = item -> {
+            latch.countDown();
+            prioritizedList.add(new String(item.getBody()));
+        };
+
+        // 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.start();
+
+        // then:
+        latch.await(10, TimeUnit.SECONDS);
+        Assert.assertEquals(2, prioritizedList.size());
+        Assert.assertEquals("2", prioritizedList.get(0));
+        Assert.assertEquals("1", prioritizedList.get(1));
+    }
+
     @Test
     public void should_enqueue_for_rabbit_queue() throws Throwable {
         // given: queue listener
@@ -109,4 +135,13 @@ public void should_enqueue_for_rabbit_queue() throws Throwable {
         Assert.assertEquals(0, rabbitQueue.size());
     }
 
+    @After
+    public void clean() {
+        inMemoryQueue.stop();
+        inMemoryQueue.clean();
+        inMemoryQueue.cleanListener();
+
+        rabbitQueue.stop();
+    }
+
 }