DefaultInMemoryStorageTest test added on Poll before improvement

common/queue/src/test/java/org/thingsboard/server/queue/memory/DefaultInMemoryStorageTest.java

@@ -15,12 +15,20 @@
  */
 package org.thingsboard.server.queue.memory;
 
+import com.google.gson.Gson;
+import lombok.extern.slf4j.Slf4j;
 import org.junit.Test;
 import org.thingsboard.server.queue.TbQueueMsg;
+import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
+@Slf4j
 public class DefaultInMemoryStorageTest {
 
     InMemoryStorage storage = new DefaultInMemoryStorage();
@@ -37,4 +45,21 @@ public class DefaultInMemoryStorageTest {
         storage.get("main");
         assertThat(storage.getLagTotal()).isEqualTo(1);
     }
-}
+
+    @Test
+    public void givenQueue_whenPoll_thenReturnList() throws InterruptedException {
+        Gson gson = new Gson();
+        String topic = "tb_core_notification.tb-node-0";
+        List<TbQueueMsg> msgs = new ArrayList<>(1001);
+        for (int i = 0; i < 1001; i++) {
+            DefaultTbQueueMsg msg = gson.fromJson("{\"key\": \"" + UUID.randomUUID() + "\"}", DefaultTbQueueMsg.class);
+            msgs.add(msg);
+            storage.put(topic, msg);
+        }
+
+        assertThat(storage.getLagTotal()).as("total lag is 1001").isEqualTo(1001);
+        assertThat(storage.get(topic)).as("poll exactly 1000 msgs").isEqualTo(msgs.subList(0, 1000));
+        assertThat(storage.get(topic)).as("poll last 1 message").isEqualTo(msgs.subList(1000, 1001));
+        assertThat(storage.getLagTotal()).as("total lag is zero").isEqualTo(0);
+    }
+}