Fixes (#101)

Author: sonus21

.circleci/config.yml

@@ -173,7 +173,7 @@ integration_test_steps: &integration_test_steps
   steps:
     - *attach_workspace
     - *redis_dep
-    - run: ./gradlew codeCoverageReport -DincludeTags=integration -DexcludeTags=redisCluster,producerOnly
+    - run: ./gradlew codeCoverageReport -DincludeTags=integration -DexcludeTags=redisCluster,producerOnly,local
     - *persist_to_workspace
     - *copy_logs
     - *copy_test_results

CHANGELOG.md

@@ -1,11 +1,14 @@
 # [Rqueue] New and Notable Changes
 
-### [2.8.1] - 19-Jul-2021
-
-Option to add rqueue web url prefix, the prefix is configured from application.properties file using
-`rqueue.web.url.prefix=my-application`, now rqueue dashboard would be served
-at  `my-application/rquque` instead of `/rqueue`, the configuration has higher priority than the
+### [2.9.0] - TBD
+### Fixes
+* Option to add rqueue web url prefix, the prefix is configured from application.properties file using
+`rqueue.web.url.prefix=/my-application/`, now rqueue dashboard would be served
+at  `/my-application/rquque` instead of `/rqueue`, the configuration has higher priority than the
 HTTP request header `x-forwarded-prefix`.
+  
+* Custom message converter is not working
+* RedisCommandExecutionException : command arguments must be strings or integers
 
 ### [2.8.0] - 08-Jun-2021
 
@@ -251,4 +254,4 @@ Fixes:
 
 [2.8.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.8.0-RELEASE
 
-[2.8.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.8.1-RELEASE
+[2.9.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.9.0-RELEASE

build.gradle

@@ -70,7 +70,7 @@ ext {
 
 subprojects {
     group = 'com.github.sonus21'
-    version = '2.8.1-RELEASE'
+    version = '2.9.0-RELEASE'
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueListenerBaseConfig.java

@@ -21,6 +21,7 @@
 import com.github.sonus21.rqueue.common.RqueueLockManager;
 import com.github.sonus21.rqueue.common.RqueueRedisTemplate;
 import com.github.sonus21.rqueue.common.impl.RqueueLockManagerImpl;
+import com.github.sonus21.rqueue.converter.MessageConverterProvider;
 import com.github.sonus21.rqueue.core.DelayedMessageScheduler;
 import com.github.sonus21.rqueue.core.ProcessingMessageScheduler;
 import com.github.sonus21.rqueue.core.RqueueBeanProvider;
@@ -66,7 +67,33 @@ public abstract class RqueueListenerBaseConfig {
   public static final int MAX_DB_VERSION = 2;
   private static final String TEMPLATE_DIR = "templates/rqueue/";
   private static final String TEMPLATE_SUFFIX = ".html";
-  protected @Value("${rqueue.reactive.enabled:false}") boolean reactiveEnabled;
+
+  @Value("${rqueue.reactive.enabled:false}")
+  protected boolean reactiveEnabled;
+
+  @Value(
+      "${rqueue.message.converter.provider.class:com.github.sonus21.rqueue.converter.DefaultMessageConverterProvider}")
+  private String messageConverterProviderClass;
+
+  protected MessageConverterProvider getMessageConverterProvider() {
+    try {
+      Class<?> c =
+          Thread.currentThread().getContextClassLoader().loadClass(messageConverterProviderClass);
+      Object messageProvider = c.newInstance();
+      if (messageProvider instanceof MessageConverterProvider) {
+        return (MessageConverterProvider) messageProvider;
+      }
+      throw new IllegalStateException(
+          "configured message converter is not of type MessageConverterProvider, type: '"
+              + messageConverterProviderClass
+              + "'",
+          new Exception());
+    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
+      throw new IllegalStateException(
+          "MessageConverterProvider class '" + messageConverterProviderClass + "' loading failed ",
+          e);
+    }
+  }
 
   @Autowired(required = false)
   protected final SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory =

rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueWebConfig.java

@@ -40,8 +40,7 @@ public class RqueueWebConfig {
 
   /**
    * Whether queue stats should be collected or not. When this flag is disabled, metric data won't
-   * be available in the dashboard. This consumes heavy CPU resources as well due to statistics
-   * aggregation and computations.
+   * be available in the dashboard.
    */
   @Value("${rqueue.web.collect.listener.stats:true}")
   private boolean collectListenerStats;
@@ -53,12 +52,19 @@ public class RqueueWebConfig {
   @Value("${rqueue.web.statistic.history.day:90}")
   private int historyDay;
 
+  // number of jobs that should be aggregated at once
   @Value("${rqueue.web.collect.statistic.aggregate.event.count:500}")
   private int aggregateEventCount;
 
+  // controls how long to wait before doing aggregation
   @Value("${rqueue.web.collect.statistic.aggregate.event.wait.time:60}")
-  private int aggregateEventWaitTime;
+  private int aggregateEventWaitTimeInSecond;
 
+  // controls how long, application should wait for the running threads to complete the execution
   @Value("${rqueue.web.collect.statistic.aggregate.shutdown.wait.time:500}")
   private int aggregateShutdownWaitTime;
+
+  // lock duration for aggregate job, acquired  per queue
+  @Value("${rqueue.web.collect.statistic.aggregate.event.lock.duration:500}")
+  private int aggregateEventLockDurationInMs;
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/config/SimpleRqueueListenerContainerFactory.java

@@ -21,7 +21,8 @@
 
 import com.github.sonus21.rqueue.annotation.MessageListener;
 import com.github.sonus21.rqueue.annotation.RqueueListener;
-import com.github.sonus21.rqueue.core.DefaultRqueueMessageConverter;
+import com.github.sonus21.rqueue.converter.DefaultMessageConverterProvider;
+import com.github.sonus21.rqueue.converter.MessageConverterProvider;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.impl.RqueueMessageTemplateImpl;
 import com.github.sonus21.rqueue.core.middleware.Middleware;
@@ -31,15 +32,13 @@
 import com.github.sonus21.rqueue.models.enums.PriorityMode;
 import com.github.sonus21.rqueue.utils.Constants;
 import com.github.sonus21.rqueue.utils.backoff.TaskExecutionBackOff;
-import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.CollectionUtils;
 
@@ -52,6 +51,9 @@
 @SuppressWarnings("WeakerAccess")
 public class SimpleRqueueListenerContainerFactory {
 
+  // The message converter provider that will return a message converter to convert messages to/from
+  private MessageConverterProvider messageConverterProvider;
+
   // Provide task executor, this can be used to provide some additional details like some threads
   // name, etc otherwise a default task executor would be created
   private AsyncTaskExecutor taskExecutor;
@@ -63,8 +65,7 @@ public class SimpleRqueueListenerContainerFactory {
   private ReactiveRedisConnectionFactory reactiveRedisConnectionFactory;
   // Custom requeue message handler
   private RqueueMessageHandler rqueueMessageHandler;
-  // The message converter to convert messages to/from
-  private MessageConverter messageConverter = new DefaultRqueueMessageConverter();
+
   // Send message poll time when no messages are available
   private long pollingInterval = 200L;
   // In case of failure how much time, we should wait for next job
@@ -90,7 +91,8 @@ public class SimpleRqueueListenerContainerFactory {
 
   private final List<Middleware> middlewares = new LinkedList<>();
 
-  // Any message headers that should be set, only used for message serialization
+  // Any message headers that should be set, headers are NOT STORED in db so it should not be
+  // changed, same header is used in serialized and deserialization process.
   private MessageHeaders messageHeaders;
 
   // Set priority mode for the pollers
@@ -158,11 +160,19 @@ public void setAutoStartup(boolean autoStartup) {
   /**
    * Return configured message handler
    *
+   * @param messageConverterProvider message converter that will be used to serialize/deserialize
+   *     message
    * @return RqueueMessageHandler object
    */
-  public RqueueMessageHandler getRqueueMessageHandler() {
+  public RqueueMessageHandler getRqueueMessageHandler(
+      MessageConverterProvider messageConverterProvider) {
+    if (this.messageConverterProvider == null) {
+      this.messageConverterProvider = messageConverterProvider;
+    }
+    notNull(this.messageConverterProvider, "messageConverterProvider can not be null");
     if (rqueueMessageHandler == null) {
-      rqueueMessageHandler = new RqueueMessageHandler(getMessageConverter(), inspectAllBean);
+      rqueueMessageHandler =
+          new RqueueMessageHandler(messageConverterProvider.getConverter(), inspectAllBean);
     }
     return rqueueMessageHandler;
   }
@@ -227,46 +237,27 @@ public void setMaxNumWorkers(int maxNumWorkers) {
   }
 
   /**
-   * @return the message converters
-   * @deprecated use {@link #getMessageConverter()}
-   */
-  @Deprecated
-  public List<MessageConverter> getMessageConverters() {
-    return Collections.singletonList(getMessageConverter());
-  }
-
-  /**
-   * For message (de)serialization we might need one or more message converters, configure those
-   * message converters
+   * Message converter must be configured before calling this method.
    *
-   * @param messageConverters list of message converters
-   * @deprecated use {@link #setMessageConverter(MessageConverter)}
+   * @throws IllegalAccessException when messageConverterProvider is null
+   * @return the message converter object
    */
-  @Deprecated
-  public void setMessageConverters(List<MessageConverter> messageConverters) {
-    notEmpty(messageConverters, "messageConverters must not be empty");
-    if (messageConverters.size() == 1) {
-      setMessageConverter(messageConverters.get(0));
-    } else {
-      setMessageConverter(new CompositeMessageConverter(messageConverters));
+  public MessageConverter getMessageConverter() throws IllegalAccessException {
+    if (messageConverterProvider == null) {
+      throw new IllegalAccessException("messageConverterProvider is not configured");
     }
-  }
-
-  /** @return the message converter */
-  public MessageConverter getMessageConverter() {
-    return messageConverter;
+    return messageConverterProvider.getConverter();
   }
 
   /**
-   * A default message converter {@link DefaultRqueueMessageConverter} is added that can handle all
-   * type of data serialization/deserialization and all data is serialized to and from JSON. You can
-   * use other mechanism to serialize/deserialize class object like MessagePack or any other format.
+   * Rqueue configures a default message converter that can convert message to/from for any object.
    *
-   * @param messageConverter the message converter
+   * @see DefaultMessageConverterProvider
+   * @param messageConverterProvider the message converter provider
    */
-  public void setMessageConverter(MessageConverter messageConverter) {
-    notNull(messageConverter, "message converter must not be null");
-    this.messageConverter = messageConverter;
+  public void setMessageConverterProvider(MessageConverterProvider messageConverterProvider) {
+    notNull(messageConverterProvider, "messageConverterProvider must not be null");
+    this.messageConverterProvider = messageConverterProvider;
   }
 
   /** @return get Redis connection factor */
@@ -308,15 +299,16 @@ public void setRqueueMessageTemplate(RqueueMessageTemplate messageTemplate) {
    * @return an object of {@link RqueueMessageListenerContainer} object
    */
   public RqueueMessageListenerContainer createMessageListenerContainer() {
-    notNull(getRqueueMessageHandler(), "rqueueMessageHandler must not be null");
     notNull(redisConnectionFactory, "redisConnectionFactory must not be null");
+    notNull(messageConverterProvider, "messageConverterProvider must not be null");
     if (rqueueMessageTemplate == null) {
       rqueueMessageTemplate =
           new RqueueMessageTemplateImpl(
               getRedisConnectionFactory(), getReactiveRedisConnectionFactory());
     }
     RqueueMessageListenerContainer messageListenerContainer =
-        new RqueueMessageListenerContainer(getRqueueMessageHandler(), rqueueMessageTemplate);
+        new RqueueMessageListenerContainer(
+            getRqueueMessageHandler(messageConverterProvider), rqueueMessageTemplate);
     messageListenerContainer.setAutoStartup(autoStartup);
     if (taskExecutor != null) {
       messageListenerContainer.setTaskExecutor(taskExecutor);
@@ -352,6 +344,9 @@ public RqueueMessageListenerContainer createMessageListenerContainer() {
     if (!CollectionUtils.isEmpty(getMiddlewares())) {
       messageListenerContainer.setMiddlewares(getMiddlewares());
     }
+    if (messageHeaders != null) {
+      messageListenerContainer.setMessageHeaders(messageHeaders);
+    }
     return messageListenerContainer;
   }
 

rqueue-core/src/main/java/com/github/sonus21/rqueue/converter/DefaultMessageConverterProvider.java

@@ -0,0 +1,28 @@
+/*
+ *  Copyright 2021 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.
+ *
+ */
+
+package com.github.sonus21.rqueue.converter;
+
+import com.github.sonus21.rqueue.core.DefaultRqueueMessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+
+public class DefaultMessageConverterProvider implements MessageConverterProvider {
+
+  @Override
+  public MessageConverter getConverter() {
+    return new DefaultRqueueMessageConverter();
+  }
+}

rqueue-core/src/main/java/com/github/sonus21/rqueue/converter/GenericMessageConverter.java

@@ -66,7 +66,7 @@ public GenericMessageConverter(ObjectMapper objectMapper) {
    */
   @Override
   public Message<?> toMessage(Object payload, MessageHeaders headers) {
-    log.debug("Payload: {} headers: {}", payload, headers);
+    log.trace("Payload: {} headers: {}", payload, headers);
     String message = smartMessageSerDes.serialize(payload);
     if (message == null) {
       return null;
@@ -76,13 +76,13 @@ public Message<?> toMessage(Object payload, MessageHeaders headers) {
 
   @Override
   public Object fromMessage(Message<?> message, Class<?> targetClass, Object conversionHint) {
-    log.debug("Message: {} class: {} hint: {}", message, targetClass, conversionHint);
+    log.trace("Message: {} class: {} hint: {}", message, targetClass, conversionHint);
     return fromMessage(message, targetClass);
   }
 
   @Override
   public Message<?> toMessage(Object payload, MessageHeaders headers, Object conversionHint) {
-    log.debug("Payload: {} headers: {} hint: {}", payload, headers, conversionHint);
+    log.trace("Payload: {} headers: {} hint: {}", payload, headers, conversionHint);
     return toMessage(payload, headers);
   }
   /**
@@ -98,7 +98,7 @@ public Message<?> toMessage(Object payload, MessageHeaders headers, Object conve
    */
   @Override
   public Object fromMessage(Message<?> message, Class<?> targetClass) {
-    log.debug("Message: {} class: {}", message, targetClass);
+    log.trace("Message: {} class: {}", message, targetClass);
     String payload;
     try {
       payload = (String) message.getPayload();
@@ -182,7 +182,7 @@ public Object deserialize(String payload) {
           return objectMapper.readValue(msg.msg, type);
         }
       } catch (Exception e) {
-        log.warn("Deserialization of message {} failed", payload, e);
+        log.debug("Deserialization of message {} failed", payload, e);
       }
       return null;
     }
@@ -194,7 +194,7 @@ public <T> T deserialize(byte[] payload, Class<T> clazz) {
       try {
         return objectMapper.readValue(payload, clazz);
       } catch (Exception e) {
-        log.warn("Deserialization of message {} failed", new String(payload), e);
+        log.debug("Deserialization of message {} failed", new String(payload), e);
       }
       return null;
     }
@@ -209,7 +209,7 @@ public String serialize(Object payload) {
         Msg message = new Msg(msg, name);
         return objectMapper.writeValueAsString(message);
       } catch (JsonProcessingException e) {
-        log.warn("Serialisation failed", e);
+        log.debug("Serialisation failed", e);
         return null;
       }
     }