Bug fixes introduced by #40 (#41)

Bug fixes introduced by #40

Author: sonus21

build.gradle

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

rqueue-common-test/src/main/java/com/github/sonus21/rqueue/test/application/CustomMessageConverterApplication.java

@@ -23,7 +23,9 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
-import org.springframework.messaging.converter.ByteArrayMessageConverter;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MappingJackson2MessageConverter;
+import org.springframework.util.MimeType;
 
 public abstract class CustomMessageConverterApplication extends ApplicationBasicConfiguration {
   @PostConstruct
@@ -48,7 +50,12 @@ public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory
         new SimpleRqueueListenerContainerFactory();
     simpleRqueueListenerContainerFactory.setRedisConnectionFactory(redisConnectionFactory);
     simpleRqueueListenerContainerFactory.setMessageConverters(
-        Collections.singletonList(new ByteArrayMessageConverter()));
+        Collections.singletonList(new MappingJackson2MessageConverter()));
+    MessageHeaders messageHeaders =
+        new MessageHeaders(
+            Collections.singletonMap(
+                MessageHeaders.CONTENT_TYPE, new MimeType("application", "json")));
+    simpleRqueueListenerContainerFactory.setMessageHeaders(messageHeaders);
     return simpleRqueueListenerContainerFactory;
   }
 }

rqueue-common-test/src/main/java/com/github/sonus21/rqueue/test/common/SpringTestBase.java

@@ -26,7 +26,7 @@
 import com.github.sonus21.rqueue.core.RqueueMessageManager;
 import com.github.sonus21.rqueue.core.RqueueMessageSender;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
-import com.github.sonus21.rqueue.core.support.RqueueMessageFactory;
+import com.github.sonus21.rqueue.core.support.RqueueMessageUtils;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer;
 import com.github.sonus21.rqueue.test.service.ConsumedMessageService;
@@ -103,17 +103,17 @@ public abstract class SpringTestBase extends TestBase {
 
   protected void enqueue(Object message, String queueName) {
     RqueueMessage rqueueMessage =
-        RqueueMessageFactory.buildMessage(
-            rqueueMessageManager.getMessageConverter(), message, queueName, null, null);
+        RqueueMessageUtils.buildMessage(
+            rqueueMessageManager.getMessageConverter(), message, queueName, null, null, null);
     rqueueMessageTemplate.addMessage(queueName, rqueueMessage);
   }
 
   protected void enqueue(String queueName, Factory factory, int n) {
     for (int i = 0; i < n; i++) {
       Object object = factory.next(i);
       RqueueMessage rqueueMessage =
-          RqueueMessageFactory.buildMessage(
-              rqueueMessageManager.getMessageConverter(), object, queueName, null, null);
+          RqueueMessageUtils.buildMessage(
+              rqueueMessageManager.getMessageConverter(), object, queueName, null, null, null);
       rqueueMessageTemplate.addMessage(queueName, rqueueMessage);
     }
   }
@@ -123,16 +123,16 @@ protected void enqueueIn(String zsetName, Factory factory, Delay delay, int n) {
       Object object = factory.next(i);
       long score = delay.getDelay(i);
       RqueueMessage rqueueMessage =
-          RqueueMessageFactory.buildMessage(
-              rqueueMessageManager.getMessageConverter(), object, zsetName, null, score);
+          RqueueMessageUtils.buildMessage(
+              rqueueMessageManager.getMessageConverter(), object, zsetName, null, score, null);
       rqueueMessageTemplate.addToZset(zsetName, rqueueMessage, rqueueMessage.getProcessAt());
     }
   }
 
   protected void enqueueIn(Object message, String zsetName, long delay) {
     RqueueMessage rqueueMessage =
-        RqueueMessageFactory.buildMessage(
-            rqueueMessageManager.getMessageConverter(), message, zsetName, null, delay);
+        RqueueMessageUtils.buildMessage(
+            rqueueMessageManager.getMessageConverter(), message, zsetName, null, delay, null);
     rqueueMessageTemplate.addToZset(zsetName, rqueueMessage, rqueueMessage.getProcessAt());
   }
 
@@ -181,6 +181,10 @@ protected void printQueueStats(List<String> queueNames) {
     }
   }
 
+  protected void printQueueStats(String queueName) {
+    printQueueStats(Collections.singletonList(queueName));
+  }
+
   protected void cleanQueue(String queue) {
     QueueDetail queueDetail = EndpointRegistry.get(queue);
     stringRqueueRedisTemplate.delete(queueDetail.getQueueName());

rqueue-common-test/src/main/java/com/github/sonus21/rqueue/test/tests/MessageChannelTests.java

@@ -33,7 +33,6 @@
 
 @Slf4j
 public abstract class MessageChannelTests extends SpringTestBase {
-  private final int messageCount = 200;
   /**
    * This test verifies whether any pending message in the delayed queue are moved or not whenever a
    * delayed message is pushed. During enqueue of delayed message we check whether there are any
@@ -42,18 +41,23 @@ public abstract class MessageChannelTests extends SpringTestBase {
    */
   protected void verifyPublishMessageIsTriggeredOnMessageAddition() throws TimedOutException {
     String delayedQueueName = rqueueConfig.getDelayedQueueName(emailQueue);
-    enqueueIn(delayedQueueName, i -> Email.newInstance(), i -> -1000L, messageCount);
+    enqueueIn(delayedQueueName, i -> Email.newInstance(), i -> -1000L, 200);
     Email email = Email.newInstance();
     log.info("adding new message {}", email);
     enqueueIn(emailQueue, email, Duration.ofMillis(1000));
     waitFor(
         () -> stringRqueueRedisTemplate.getZsetSize(delayedQueueName) <= 1,
         "one or zero messages in zset");
     assertTrue(
-        stringRqueueRedisTemplate.getListSize(rqueueConfig.getQueueName(emailQueue))
-            >= messageCount,
+        stringRqueueRedisTemplate.getListSize(rqueueConfig.getQueueName(emailQueue)) >= 200,
         "Messages are correctly moved");
-    assertEquals(messageCount + 1L, getMessageCount(emailQueue));
+    assertEquals(
+        200 + 1L,
+        getMessageCount(emailQueue),
+        () -> {
+          printQueueStats(emailQueue);
+          return "message count is not correct";
+        });
   }
 
   /**
@@ -67,7 +71,7 @@ protected void verifyPublishMessageIsTriggeredOnMessageRemoval() throws TimedOut
     List<String> ids = new ArrayList<>();
     int maxDelay = 2000;
     String processingQueue = rqueueConfig.getProcessingQueueName(jobQueue);
-    for (int i = 0; i < messageCount; i++) {
+    for (int i = 0; i < 200; i++) {
       Job job = Job.newInstance();
       jobs.add(job);
       ids.add(job.getId());
@@ -79,11 +83,9 @@ protected void verifyPublishMessageIsTriggeredOnMessageRemoval() throws TimedOut
     }
     TimeoutUtils.sleep(maxDelay);
     waitFor(
-        () -> 0 == getMessageCount(jobQueue),
-        30 * Constants.ONE_MILLI,
-        "messages to be consumed");
+        () -> 0 == getMessageCount(jobQueue), 30 * Constants.ONE_MILLI, "messages to be consumed");
     waitFor(
-        () -> messageCount == consumedMessageService.getMessages(ids, Job.class).size(),
+        () -> 200 == consumedMessageService.getMessages(ids, Job.class).size(),
         30 * Constants.ONE_MILLI,
         "message count to be matched");
     waitFor(

rqueue-common-test/src/main/resources/producer-only-application.properties

@@ -0,0 +1,46 @@
+spring.redis.host=localhost
+spring.redis.port=8012
+job.queue.name=job-queue
+job.queue.active=true
+notification.queue.name=notification-queue
+notification.queue.delay=1000
+notification.queue.retry.count=3
+notification.queue.active=true
+email.queue.name=email-queue
+email.dead.letter.queue.name=email-dlq
+email.queue.retry.count=3
+email.queue.active=true
+mysql.db.name=test
+rqueue.metrics.tags.rqueue=test
+email.execution.time=15*60*1000
+sms.queue=sms
+sms.queue.active=false
+sms.queue.group=
+sms.queue.priority=critical:10, high:6, medium:4, low:2
+sms.queue.concurrency=-1
+chat.indexing.queue=chat-indexing
+chat.indexing.queue.active=false
+chat.indexing.queue.priority=30
+chat.indexing.queue.group=test-group
+chat.indexing.queue.concurrency=-1
+feed.generation.queue=feed-generation
+feed.generation.queue.active=false
+feed.generation.queue.priority=20
+feed.generation.queue.group=test-group
+feed.generation.queue.concurrency=-1
+reservation.queue=reservation
+reservation.queue.active=false
+reservation.queue.priority=10
+reservation.queue.group=test-group
+reservation.queue.concurrency=-1
+reservation.request.queue.name=reservation-request
+reservation.request.dead.letter.queue.name=reservation-request-dlq
+reservation.request.queue.retry.count=2
+reservation.request.dead.letter.consumer.enabled=false
+reservation.request.active=false
+reservation.request.dead.letter.queue.retry.count=1
+list.email.queue.name=email-list-queue
+list.email.queue.enabled=false
+rqueue.system.mode=PRODUCER
+
+

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

@@ -34,6 +34,7 @@
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 
@@ -55,7 +56,7 @@ public class SimpleRqueueListenerContainerFactory {
   // Custom requeue message handler
   private RqueueMessageHandler rqueueMessageHandler;
   // The message converter to convert messages to/from
-  private MessageConverter messageConverter;
+  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
@@ -79,6 +80,9 @@ public class SimpleRqueueListenerContainerFactory {
   // Any custom message requeue message template.
   private RqueueMessageTemplate rqueueMessageTemplate;
 
+  // Any message headers that should be set, only used for message serialization
+  private MessageHeaders messageHeaders;
+
   // Set priority mode for the pollers
   private PriorityMode priorityMode;
 
@@ -205,7 +209,8 @@ public void setMaxNumWorkers(int maxNumWorkers) {
     this.maxNumWorkers = maxNumWorkers;
   }
 
-  /** @return the message converters
+  /**
+   * @return the message converters
    * @deprecated use {@link #getMessageConverter()}
    */
   @Deprecated
@@ -215,9 +220,6 @@ public List<MessageConverter> getMessageConverters() {
 
   /** @return the message converter */
   public MessageConverter getMessageConverter() {
-    if (messageConverter == null) {
-      messageConverter = new DefaultRqueueMessageConverter();
-    }
     return messageConverter;
   }
 
@@ -239,6 +241,10 @@ public void setMessageConverters(List<MessageConverter> messageConverters) {
   }
 
   /**
+   * 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.
+   *
    * @param messageConverter the message converter
    */
   public void setMessageConverter(MessageConverter messageConverter) {
@@ -452,4 +458,13 @@ public PriorityMode getPriorityMode() {
   public void setPriorityMode(PriorityMode priorityMode) {
     this.priorityMode = priorityMode;
   }
+
+  public MessageHeaders getMessageHeaders() {
+    return messageHeaders;
+  }
+
+  public void setMessageHeaders(MessageHeaders messageHeaders) {
+    notEmpty(messageHeaders, "messageHeaders can not be empty");
+    this.messageHeaders = messageHeaders;
+  }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/DefaultRqueueMessageConverter.java

@@ -17,17 +17,13 @@
 package com.github.sonus21.rqueue.core;
 
 import com.github.sonus21.rqueue.converter.GenericMessageConverter;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
 import com.google.common.collect.ImmutableList;
 import lombok.EqualsAndHashCode;
 import org.springframework.messaging.converter.CompositeMessageConverter;
-import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.StringMessageConverter;
 
 @EqualsAndHashCode(callSuper = true)
-public final class DefaultRqueueMessageConverter extends CompositeMessageConverter implements MessageConverter{
+public final class DefaultRqueueMessageConverter extends CompositeMessageConverter {
 
   public DefaultRqueueMessageConverter() {
     super(ImmutableList.of(new GenericMessageConverter(), new StringMessageConverter()));

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -16,8 +16,8 @@
 
 package com.github.sonus21.rqueue.core.impl;
 
+import static com.github.sonus21.rqueue.core.support.RqueueMessageUtils.buildMessage;
 import static com.github.sonus21.rqueue.utils.Constants.MIN_DELAY;
-import static com.github.sonus21.rqueue.utils.MessageUtils.buildMessage;
 import static com.github.sonus21.rqueue.utils.Validator.validateQueue;
 import static org.springframework.util.Assert.notNull;
 
@@ -34,6 +34,7 @@
 import java.time.Duration;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.MessageConverter;
 
 @Slf4j
@@ -44,13 +45,17 @@ abstract class BaseMessageSender {
   @Autowired protected RqueueRedisTemplate<String> stringRqueueRedisTemplate;
   @Autowired protected RqueueConfig rqueueConfig;
   @Autowired protected RqueueMessageMetadataService rqueueMessageMetadataService;
+  protected final MessageHeaders messageHeaders;
 
   BaseMessageSender(
-      RqueueMessageTemplate messageTemplate, MessageConverter messageConverter) {
+      RqueueMessageTemplate messageTemplate,
+      MessageConverter messageConverter,
+      MessageHeaders messageHeaders) {
     notNull(messageTemplate, "messageTemplate cannot be null");
     notNull(messageConverter, "messageConverter cannot be null");
     this.messageTemplate = messageTemplate;
     this.messageConverter = messageConverter;
+    this.messageHeaders = messageHeaders;
   }
 
   private void storeMessageMetadata(RqueueMessage rqueueMessage, Long delayInMillis) {
@@ -75,7 +80,8 @@ private RqueueMessage constructMessage(
       Integer retryCount,
       Long delayInMilliSecs) {
     RqueueMessage rqueueMessage =
-        buildMessage(messageConverter, queueName, message, retryCount, delayInMilliSecs);
+        buildMessage(
+            messageConverter, message, queueName, retryCount, delayInMilliSecs, messageHeaders);
     if (messageId != null) {
       rqueueMessage.setId(messageId);
     }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueEndpointManagerImpl.java

@@ -28,14 +28,17 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.CollectionUtils;
 
 public class RqueueEndpointManagerImpl extends BaseMessageSender implements RqueueEndpointManager {
 
-  public RqueueEndpointManagerImpl(RqueueMessageTemplate messageTemplate,
-      MessageConverter messageConverter) {
-    super(messageTemplate, messageConverter);
+  public RqueueEndpointManagerImpl(
+      RqueueMessageTemplate messageTemplate,
+      MessageConverter messageConverter,
+      MessageHeaders messageHeaders) {
+    super(messageTemplate, messageConverter, messageHeaders);
   }
 
   @Override

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageEnqueuerImpl.java

@@ -26,17 +26,18 @@
 import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.utils.PriorityUtils;
-import java.util.Collections;
-import java.util.List;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.MessageConverter;
 
 @Slf4j
 public class RqueueMessageEnqueuerImpl extends BaseMessageSender implements RqueueMessageEnqueuer {
 
   public RqueueMessageEnqueuerImpl(
-      RqueueMessageTemplate messageTemplate, MessageConverter messageConverter) {
-    super(messageTemplate, messageConverter);
+      RqueueMessageTemplate messageTemplate,
+      MessageConverter messageConverter,
+      MessageHeaders messageHeaders) {
+    super(messageTemplate, messageConverter, messageHeaders);
   }
 
   @Override