More test cases.

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/ProcessingMessageSchedulerTest.java

@@ -53,9 +53,10 @@
 @TestPropertySource(
     properties = {
       "rqueue.scheduler.auto.start=false",
-      "spring.redis.port=6383",
+      "spring.redis.port=6379",
       "mysql.db.name=test3",
-      "max.workers.count=120"
+      "max.workers.count=120",
+      "use.system.redis=true"
     })
 public class ProcessingMessageSchedulerTest {
   static {
@@ -74,7 +75,7 @@ public class ProcessingMessageSchedulerTest {
   public RunTestUntilFail retry =
       new RunTestUntilFail(
           log,
-          1,
+          3,
           () -> {
             for (Entry<String, List<RqueueMessage>> entry :
                 TestUtils.getMessageMap(jobQueueName, rqueueMessageTemplate).entrySet()) {

rqueue-test-common/src/main/resources/logback.xml

@@ -17,7 +17,6 @@
   </appender>
 
   <root level="ERROR">
-    <appender-ref ref="STDOUT"/>
     <appender-ref ref="FILE"/>
   </root>
 

rqueue/src/main/java/com/github/sonus21/rqueue/models/MessageMoveResult.java

@@ -18,14 +18,10 @@
 
 import com.github.sonus21.rqueue.models.response.BaseResponse;
 import lombok.AllArgsConstructor;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import lombok.ToString;
 
 @Getter
 @AllArgsConstructor
-@ToString(callSuper = true)
-@EqualsAndHashCode(callSuper = true)
 public class MessageMoveResult extends BaseResponse {
   private static final long serialVersionUID = 2419344664633952274L;
   private int numberOfMessages;

rqueue/src/main/java/com/github/sonus21/rqueue/models/db/JobRunTime.java

@@ -22,14 +22,12 @@
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
-import lombok.ToString;
 
 @Getter
 @Setter
 @NoArgsConstructor
-@ToString
-@EqualsAndHashCode(callSuper = false)
 @AllArgsConstructor
+@EqualsAndHashCode(callSuper = false)
 public class JobRunTime extends SerializableBase {
   private static final long serialVersionUID = 6219118148061766036L;
   private long min;

rqueue/src/main/java/com/github/sonus21/rqueue/models/event/QueueInitializationEvent.java

@@ -23,7 +23,6 @@
 import org.springframework.context.ApplicationEvent;
 
 @Getter
-@ToString(callSuper = true)
 public class QueueInitializationEvent extends ApplicationEvent {
 
   private static final long serialVersionUID = 1955427920805054136L;

rqueue/src/main/java/com/github/sonus21/rqueue/models/request/ChartDataRequest.java

@@ -21,15 +21,13 @@
 import com.github.sonus21.rqueue.models.enums.ChartDataType;
 import com.github.sonus21.rqueue.models.enums.ChartType;
 import java.util.List;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
 
 @Getter
 @Setter
-@AllArgsConstructor
 @NoArgsConstructor
 @ToString
 public class ChartDataRequest extends SerializableBase {

rqueue/src/main/java/com/github/sonus21/rqueue/models/request/MoveMessageRequest.java

@@ -23,8 +23,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import lombok.AllArgsConstructor;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
@@ -33,9 +31,7 @@
 @Getter
 @Setter
 @NoArgsConstructor
-@AllArgsConstructor
 @ToString
-@EqualsAndHashCode(callSuper = false)
 public class MoveMessageRequest implements Serializable {
   private static final long serialVersionUID = -5105668034442269108L;
   private DataType srcType;

rqueue/src/main/java/com/github/sonus21/rqueue/models/response/DataViewResponse.java

@@ -18,6 +18,7 @@
 
 import com.github.sonus21.rqueue.models.enums.ActionType;
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
@@ -34,7 +35,7 @@
 public class DataViewResponse extends BaseResponse {
   private static final long serialVersionUID = -8359552138158857044L;
   private List<String> headers;
-  private ActionType action;
+  private List<ActionType> actions = new ArrayList<>();
   private List<List<Serializable>> rows;
 
   public static DataViewResponse createErrorMessage(String message) {
@@ -43,4 +44,8 @@ public static DataViewResponse createErrorMessage(String message) {
     response.setMessage(message);
     return response;
   }
+
+  public void addAction(ActionType actionType) {
+    actions.add(actionType);
+  }
 }

rqueue/src/main/java/com/github/sonus21/rqueue/web/controller/RqueueViewController.java

@@ -82,6 +82,7 @@ private void addBasicDetails(Model model) {
     model.addAttribute("mavenRepoLink", Constants.MAVEN_REPO_LINK);
     model.addAttribute(
         "time", OffsetDateTime.now(ZoneOffset.UTC).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
+    model.addAttribute("timeInMilli", System.currentTimeMillis());
     model.addAttribute("version", rqueueConfig.getVersion());
   }
 

rqueue/src/main/java/com/github/sonus21/rqueue/web/dao/impl/RqueueQStatsDaoImpl.java

@@ -24,23 +24,20 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
-import javax.annotation.PostConstruct;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Repository;
 
 @Repository
 public class RqueueQStatsDaoImpl implements RqueueQStatsDao {
-  private final RqueueConfig rqueueConfig;
-  private RqueueRedisTemplate<QueueStatistics> rqueueRedisTemplate;
+  private final RqueueRedisTemplate<QueueStatistics> rqueueRedisTemplate;
 
   @Autowired
   public RqueueQStatsDaoImpl(RqueueConfig rqueueConfig) {
-    this.rqueueConfig = rqueueConfig;
+    this(new RqueueRedisTemplate<>(rqueueConfig.getConnectionFactory()));
   }
 
-  @PostConstruct
-  public void init() {
-    this.rqueueRedisTemplate = new RqueueRedisTemplate<>(rqueueConfig.getConnectionFactory());
+  public RqueueQStatsDaoImpl(RqueueRedisTemplate<QueueStatistics> rqueueRedisTemplate) {
+    this.rqueueRedisTemplate = rqueueRedisTemplate;
   }
 
   @Override
@@ -57,6 +54,9 @@ public List<QueueStatistics> findAll(Collection<String> ids) {
 
   @Override
   public void save(QueueStatistics queueStatistics) {
+    if (queueStatistics == null) {
+      throw new IllegalArgumentException("queueStatistics cannot be null");
+    }
     if (queueStatistics.getId() == null) {
       throw new IllegalArgumentException("id cannot be null " + queueStatistics);
     }

rqueue/src/main/java/com/github/sonus21/rqueue/web/dao/impl/RqueueSystemConfigDaoImpl.java

@@ -37,7 +37,7 @@ public class RqueueSystemConfigDaoImpl implements RqueueSystemConfigDao {
 
   @Autowired
   public RqueueSystemConfigDaoImpl(RqueueConfig rqueueConfig) {
-    this.rqueueRedisTemplate = new RqueueRedisTemplate<>(rqueueConfig.getConnectionFactory());
+    this(new RqueueRedisTemplate<>(rqueueConfig.getConnectionFactory()));
   }
 
   public RqueueSystemConfigDaoImpl(RqueueRedisTemplate<QueueConfig> rqueueRedisTemplate) {

rqueue/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueDashboardChartServiceImpl.java

@@ -196,7 +196,7 @@ private Map<Integer, TasksStat> aggregateData(
         idToTasksStat.put(i, new TasksStat());
       }
     }
-    if (jobRunTime.getMax() == 0) {
+    if (jobRunTime.getMax() == 0 && idToTasksStat.isEmpty()) {
       idToTasksStat.put(0, new TasksStat());
     }
     for (int i = jobRunTime.getMax() + 1; i < maxRequired; i++) {
@@ -224,7 +224,7 @@ private List<Serializable> getHeader(String title, List<ChartDataType> chartData
   private List<Serializable> createRow(
       Integer id, TasksStat tasksStat, List<ChartDataType> chartDataTypeList) {
     List<Serializable> row = new LinkedList<>();
-    row.add(id);
+    row.add(String.valueOf(id + 1));
     for (ChartDataType status : chartDataTypeList) {
       switch (status) {
         case RETRIED:

rqueue/src/main/java/com/github/sonus21/rqueue/web/service/impl/RqueueQDetailServiceImpl.java

@@ -254,9 +254,9 @@ public DataViewResponse getExplorePageData(
     setHeadersIfRequired(deadLetterQueue, type, response, pageNumber);
 
     if (deadLetterQueue) {
-      response.setAction(ActionType.DELETE);
+      response.addAction(ActionType.DELETE);
     } else {
-      response.setAction(ActionType.NONE);
+      response.addAction(ActionType.NONE);
     }
     switch (type) {
       case ZSET:
@@ -284,7 +284,6 @@ private DataViewResponse responseForSet(String name) {
     List<Object> items = new ArrayList<>(stringRqueueRedisTemplate.getMembers(name));
     DataViewResponse response = new DataViewResponse();
     response.setHeaders(Collections.singletonList("Item"));
-    response.setAction(ActionType.NONE);
     List<List<Serializable>> rows = new ArrayList<>();
     for (Object item : items) {
       rows.add(Collections.singletonList(item.toString()));
@@ -296,7 +295,6 @@ private DataViewResponse responseForSet(String name) {
   private DataViewResponse responseForKeyVal(String name) {
     DataViewResponse response = new DataViewResponse();
     response.setHeaders(Collections.singletonList("Value"));
-    response.setAction(ActionType.NONE);
     Object val = stringRqueueRedisTemplate.get(name);
     List<List<Serializable>> rows =
         Collections.singletonList(Collections.singletonList(String.valueOf(val)));
@@ -307,7 +305,6 @@ private DataViewResponse responseForKeyVal(String name) {
   private DataViewResponse responseForZset(
       String name, String key, int pageNumber, int itemPerPage) {
     DataViewResponse response = new DataViewResponse();
-    response.setAction(ActionType.NONE);
     int start = pageNumber * itemPerPage;
     int end = start + itemPerPage - 1;
     List<List<Serializable>> rows = new ArrayList<>();
@@ -328,7 +325,6 @@ private DataViewResponse responseForZset(
   private DataViewResponse responseForList(String name, int pageNumber, int itemPerPage) {
     DataViewResponse response = new DataViewResponse();
     response.setHeaders(Collections.singletonList("Item"));
-    response.setAction(ActionType.NONE);
     int start = pageNumber * itemPerPage;
     int end = start + itemPerPage - 1;
     List<List<Serializable>> rows = new ArrayList<>();

rqueue/src/main/resources/public/rqueue/js/rqueue.js

@@ -131,10 +131,15 @@ function displayTable(nextOrPrev) {
       var tableHeader = $('#table-header');
       var displayPageNumberEl = $('#display-page-number');
       displayPageNumberEl.empty();
-      if (response.action === 'DELETE') {
-        $('#clear-queue').show();
-      } else {
-        $('#clear-queue').hide();
+      $('#clear-queue').hide();
+      for (var i = 0; i < response.actions.length; i++) {
+        switch (response.actions[i]) {
+          case 'DELETE':
+            $('#clear-queue').show();
+            break;
+          default:
+            console.log("Unknown action", response.actions[i])
+        }
       }
       if (rows.length === pageSize) {
         $('#next-page-button').show();
@@ -384,7 +389,7 @@ $('#move-button').on("click", function () {
   if ($('#priority-controller-form').is(":visible")) {
     var val = $('#priority-val').val();
     if (val !== '') {
-      other['newScore'] = parseInt(val);
+      other['newScore'] = val;
     }
     var type = getSelectedOption('priority-type');
     if (type === 'REL') {
@@ -396,8 +401,13 @@ $('#move-button').on("click", function () {
   payload['other'] = other;
   ajaxRequest('/rqueue/api/v1/move', 'PUT', payload,
       function (response) {
-        alert(
-            "Transferred " + response.numberOfMessageTransferred + " messages");
+        if (response.code === 0) {
+          alert(
+              "Transferred " + response.numberOfMessageTransferred
+              + " messages");
+        } else {
+          alert(response.message);
+        }
       },
       function (response) {
         console.log('failed, ' + response);

rqueue/src/main/resources/scripts/add_message.lua

@@ -1,17 +1,3 @@
--- Copyright 2020 Sonu Kumar
---
--- Licensed under the Apache License, Version 2.0 (the "License");
--- you may not use this file except in compliance with the License.
--- You may obtain a copy of the License at
---
---       https://www.apache.org/licenses/LICENSE-2.0
---
--- Unless required by applicable law or agreed to in writing, software
--- distributed under the License is distributed on an "AS IS" BASIS,
--- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
--- See the License for the specific language governing permissions and
--- limitations under the License.
-
 local count = redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
 --if elements with lower priority are on head
 local v = redis.call('ZRANGE', KEYS[1], 0, 0, 'WITHSCORES')

rqueue/src/main/resources/scripts/move_message_list_to_list.lua

@@ -1,22 +1,8 @@
--- Copyright 2020 Sonu Kumar
---
--- Licensed under the Apache License, Version 2.0 (the "License");
--- you may not use this file except in compliance with the License.
--- You may obtain a copy of the License at
---
---       https://www.apache.org/licenses/LICENSE-2.0
---
--- Unless required by applicable law or agreed to in writing, software
--- distributed under the License is distributed on an "AS IS" BASIS,
--- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
--- See the License for the specific language governing permissions and
--- limitations under the License.
-
 for i = tonumber(ARGV[1]), 1, -1
 do
-    local msg = redis.call('LRANGE', KEYS[1], 0, 0)
+    local msg = redis.call('LRANGE', KEYS[1], 0, 0)[1]
     if msg ~= nil then
-        redis.call("RPUSH", KEYS[2], msg[1])
+        redis.call("RPUSH", KEYS[2], msg)
         redis.call("LPOP", KEYS[1])
     else
         break