test: refactored for assertThat for DefaultTbQueueRequestTemplateTest

common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java

@@ -158,11 +158,10 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
 
     void setTimeoutException(UUID key, ResponseMetaData<Response> staleRequest, long currentNs) {
         if (currentNs >= staleRequest.getSubmitTime() + staleRequest.getTimeout()) {
-            log.info("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key);
+            log.warn("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key);
         } else {
             log.error("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key);
         }
-
         staleRequest.future.setException(new TimeoutException());
     }
 
@@ -218,7 +217,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
         long currentClockNs = getCurrentClockNs();
         SettableFuture<Response> future = SettableFuture.create();
         ResponseMetaData<Response> responseMetaData = new ResponseMetaData<>(currentClockNs + maxRequestTimeoutNs, future, currentClockNs, maxRequestTimeoutNs);
-        log.info("pending {}", responseMetaData); //TODO trace
+        log.trace("pending {}", responseMetaData);
         if (pendingRequests.putIfAbsent(requestId, responseMetaData) != null) {
             log.warn("Pending request already exists [{}]!", maxPendingRequests);
             return Futures.immediateFailedFuture(new RuntimeException("Pending request already exists !" + requestId));

common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java

@@ -87,7 +87,7 @@ public class DefaultTbQueueRequestTemplateTest {
     ExecutorService executor;
     String topic = "js-responses-tb-node-0";
     long maxRequestTimeout = 10;
-    long maxPendingRequests = 1000;
+    long maxPendingRequests = 32;
     long pollInterval = 5;
 
     DefaultTbQueueRequestTemplate inst;
@@ -124,14 +124,14 @@ public class DefaultTbQueueRequestTemplateTest {
 
         inst.init();
         //assertNotEquals(0, inst.tickTs);
-        assertEquals(0, inst.nextCleanupNs);
+        assertThat(inst.nextCleanupNs, equalTo(0L));
         verify(queueAdmin, times(1)).createTopicIfNotExists(topic);
         verify(requestTemplate, times(1)).init();
         verify(responseTemplate, times(1)).subscribe();
         verify(executorMock, times(1)).submit(any(Runnable.class));
 
         inst.stop();
-        assertTrue(inst.stopped);
+        assertThat(inst.stopped, is(true));
         verify(responseTemplate, times(1)).unsubscribe();
         verify(requestTemplate, times(1)).stop();
         verify(executorMock, never()).shutdownNow();
@@ -165,11 +165,11 @@ public class DefaultTbQueueRequestTemplateTest {
     public void givenMessages_whenSend_thenOK() {
         willDoNothing().given(inst).sendToRequestTemplate(any(), any(), any(), any());
         inst.init();
-        int msgCount = 10;
+        final int msgCount = 10;
         for (int i = 0; i < msgCount; i++) {
             inst.send(getRequestMsgMock());
         }
-        assertEquals(msgCount, inst.pendingRequests.mappingCount());
+        assertThat(inst.pendingRequests.mappingCount(), equalTo((long) msgCount));
         verify(inst, times(msgCount)).sendToRequestTemplate(any(), any(), any(), any());
     }
 
@@ -179,10 +179,10 @@ public class DefaultTbQueueRequestTemplateTest {
         inst.init();
         int msgOverflowCount = 10;
         for (int i = 0; i < inst.maxPendingRequests; i++) {
-            assertFalse(inst.send(getRequestMsgMock()).isDone()); //SettableFuture future - pending only
+            assertThat(inst.send(getRequestMsgMock()).isDone(), is(false)); //SettableFuture future - pending only
         }
         for (int i = 0; i < msgOverflowCount; i++) {
-            assertTrue("max pending requests overflow", inst.send(getRequestMsgMock()).isDone()); //overflow, immediate failed future
+            assertThat("max pending requests overflow", inst.send(getRequestMsgMock()).isDone(), is(true)); //overflow, immediate failed future
         }
         assertThat(inst.pendingRequests.mappingCount(), equalTo(inst.maxPendingRequests));
         verify(inst, times((int) inst.maxPendingRequests)).sendToRequestTemplate(any(), any(), any(), any());
@@ -205,7 +205,7 @@ public class DefaultTbQueueRequestTemplateTest {
         long stepNs = TimeUnit.MILLISECONDS.toNanos(1);
         for (long i = 0; i <= inst.maxRequestTimeoutNs * 2; i = i + stepNs) {
             currentTime.addAndGet(stepNs);
-            assertFalse(inst.send(getRequestMsgMock()).isDone()); //SettableFuture future - pending only
+            assertThat(inst.send(getRequestMsgMock()).isDone(), is(false)); //SettableFuture future - pending only
             if (i % (inst.maxRequestTimeoutNs * 3 / 2) == 0) {
                 inst.fetchAndProcessResponses();
             }
@@ -222,7 +222,6 @@ public class DefaultTbQueueRequestTemplateTest {
         for (int i = 0; i < responseMetaDataList.size(); i++) {
             assertThat("tickTs >= calculatedExpTime", tickTsList.get(i), greaterThanOrEqualTo(responseMetaDataList.get(i).getSubmitTime() + responseMetaDataList.get(i).getTimeout()));
         }
-
     }
 
     TbQueueMsg getRequestMsgMock() {