Added isEmpty check method to unsafe persistent queue

broker/src/main/java/io/moquette/broker/unsafequeues/Queue.java

@@ -11,7 +11,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Not thread safe disk persisted queue.S
+ * Not thread safe disk persisted queue.
  * */
 public class Queue {
     private static final Logger LOG = LoggerFactory.getLogger(Queue.class);
@@ -149,6 +149,14 @@ VirtualPointer currentTail() {
         return currentTailPtr;
     }
 
+    public boolean isEmpty() {
+        if (isTailFirstUsage(currentTailPtr)) {
+            return currentHeadPtr.compareTo(currentTailPtr) == 0;
+        } else {
+            return currentHeadPtr.moveForward(1).compareTo(currentTailPtr) == 0;
+        }
+    }
+
     /**
      * Read next message or return null if the queue has no data.
      * */

broker/src/test/java/io/moquette/broker/unsafequeues/QueueTest.java

@@ -98,6 +98,27 @@ private ByteBuffer randomPayload(int dataSize) {
         return (ByteBuffer) ByteBuffer.wrap(payload);
     }
 
+    @Test
+    public void newlyQueueIsEmpty() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+
+        assertTrue(queue.isEmpty(), "Freshly created queue must be empty");
+    }
+
+    @Test
+    public void consumedQueueIsEmpty() throws QueueException {
+        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);
+        final Queue queue = queuePool.getOrCreate("test");
+        queue.enqueue(ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));
+        Optional<ByteBuffer> data = queue.dequeue();
+        assertTrue(data.isPresent(), "Some payload is retrieved");
+
+        assertEquals(4, data.get().remaining(), "Payload contains what's expected");
+
+        assertTrue(queue.isEmpty(), "Queue must be empty after consuming it");
+    }
+
     @Test
     public void insertSomeDataIntoNewQueue() throws QueueException, IOException {
         final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder);