merged main

Author: sonus21

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

@@ -45,6 +45,7 @@ public class RqueueConfig {
 
   @Getter
   private static final String brokerId = UUID.randomUUID().toString();
+
   private static final AtomicLong counter = new AtomicLong(1);
   private final RedisConnectionFactory connectionFactory;
   private final ReactiveRedisConnectionFactory reactiveRedisConnectionFactory;

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

@@ -35,13 +35,13 @@
 import com.github.sonus21.rqueue.dao.impl.RqueueStringDaoImpl;
 import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer;
 import com.github.sonus21.rqueue.metrics.RqueueQueueMetrics;
-import com.github.sonus21.rqueue.worker.RqueueWorkerRegistry;
-import com.github.sonus21.rqueue.worker.RqueueWorkerRegistryImpl;
 import com.github.sonus21.rqueue.utils.RedisUtils;
 import com.github.sonus21.rqueue.utils.condition.MissingRqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.utils.condition.ReactiveEnabled;
 import com.github.sonus21.rqueue.utils.pebble.ResourceLoader;
 import com.github.sonus21.rqueue.utils.pebble.RqueuePebbleExtension;
+import com.github.sonus21.rqueue.worker.RqueueWorkerRegistry;
+import com.github.sonus21.rqueue.worker.RqueueWorkerRegistryImpl;
 import io.pebbletemplates.pebble.PebbleEngine;
 import io.pebbletemplates.spring.extension.SpringExtension;
 import io.pebbletemplates.spring.reactive.PebbleReactiveViewResolver;

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

@@ -24,8 +24,8 @@
 import com.github.sonus21.rqueue.dao.RqueueSystemConfigDao;
 import com.github.sonus21.rqueue.listener.RqueueMessageHandler;
 import com.github.sonus21.rqueue.metrics.RqueueMetricsCounter;
-import com.github.sonus21.rqueue.worker.RqueueWorkerRegistry;
 import com.github.sonus21.rqueue.web.service.RqueueMessageMetadataService;
+import com.github.sonus21.rqueue.worker.RqueueWorkerRegistry;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.beans.factory.annotation.Autowired;

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

@@ -53,9 +53,15 @@ abstract class BaseMessageSender {
   protected final MessageConverter messageConverter;
   protected final RqueueMessageTemplate messageTemplate;
   protected final RqueueMessageIdGenerator messageIdGenerator;
-  @Autowired protected RqueueStringDao rqueueStringDao;
-  @Autowired protected RqueueConfig rqueueConfig;
-  @Autowired protected RqueueMessageMetadataService rqueueMessageMetadataService;
+
+  @Autowired
+  protected RqueueStringDao rqueueStringDao;
+
+  @Autowired
+  protected RqueueConfig rqueueConfig;
+
+  @Autowired
+  protected RqueueMessageMetadataService rqueueMessageMetadataService;
 
   BaseMessageSender(
       RqueueMessageTemplate messageTemplate,
@@ -118,16 +124,15 @@ protected String pushMessage(
       Long delayInMilliSecs,
       boolean isUnique) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage =
-        buildMessage(
-            messageIdGenerator,
-            messageConverter,
-            queueName,
-            messageId,
-            message,
-            retryCount,
-            delayInMilliSecs,
-            messageHeaders);
+    RqueueMessage rqueueMessage = buildMessage(
+        messageIdGenerator,
+        messageConverter,
+        queueName,
+        messageId,
+        message,
+        retryCount,
+        delayInMilliSecs,
+        messageHeaders);
     try {
       storeMessageMetadata(rqueueMessage, delayInMilliSecs, false, isUnique);
       enqueue(queueDetail, rqueueMessage, delayInMilliSecs, false);
@@ -147,16 +152,15 @@ protected String pushMessage(
   protected String pushPeriodicMessage(
       String queueName, String messageId, Object message, long periodInMilliSeconds) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage =
-        buildPeriodicMessage(
-            messageIdGenerator,
-            messageConverter,
-            queueName,
-            messageId,
-            message,
-            null,
-            periodInMilliSeconds,
-            messageHeaders);
+    RqueueMessage rqueueMessage = buildPeriodicMessage(
+        messageIdGenerator,
+        messageConverter,
+        queueName,
+        messageId,
+        message,
+        null,
+        periodInMilliSeconds,
+        messageHeaders);
     try {
       storeMessageMetadata(rqueueMessage, periodInMilliSeconds, false, false);
       enqueue(queueDetail, rqueueMessage, periodInMilliSeconds, false);

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

@@ -42,11 +42,7 @@ public ReactiveRqueueMessageEnqueuerImpl(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
       MessageHeaders messageHeaders) {
-    this(
-        messageTemplate,
-        messageConverter,
-        messageHeaders,
-        new UuidV4RqueueMessageIdGenerator());
+    this(messageTemplate, messageConverter, messageHeaders, new UuidV4RqueueMessageIdGenerator());
   }
 
   public ReactiveRqueueMessageEnqueuerImpl(
@@ -68,16 +64,15 @@ private <T> Mono<T> pushReactiveMessage(
       boolean isUnique,
       Function<RqueueMessage, Mono<T>> monoConverter) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage =
-        builder.build(
-            messageIdGenerator,
-            messageConverter,
-            queueName,
-            messageId,
-            message,
-            retryCount,
-            delayInMilliSecs,
-            messageHeaders);
+    RqueueMessage rqueueMessage = builder.build(
+        messageIdGenerator,
+        messageConverter,
+        queueName,
+        messageId,
+        message,
+        retryCount,
+        delayInMilliSecs,
+        messageHeaders);
     try {
       Mono<Boolean> storeResult =
           (Mono<Boolean>) storeMessageMetadata(rqueueMessage, delayInMilliSecs, true, isUnique);

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

@@ -52,11 +52,7 @@ public RqueueEndpointManagerImpl(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
       MessageHeaders messageHeaders) {
-    this(
-        messageTemplate,
-        messageConverter,
-        messageHeaders,
-        new UuidV4RqueueMessageIdGenerator());
+    this(messageTemplate, messageConverter, messageHeaders, new UuidV4RqueueMessageIdGenerator());
   }
 
   public RqueueEndpointManagerImpl(

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

@@ -40,11 +40,7 @@ public RqueueMessageEnqueuerImpl(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
       MessageHeaders messageHeaders) {
-    this(
-        messageTemplate,
-        messageConverter,
-        messageHeaders,
-        new UuidV4RqueueMessageIdGenerator());
+    this(messageTemplate, messageConverter, messageHeaders, new UuidV4RqueueMessageIdGenerator());
   }
 
   public RqueueMessageEnqueuerImpl(

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

@@ -51,11 +51,7 @@ public RqueueMessageManagerImpl(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
       MessageHeaders messageHeaders) {
-    this(
-        messageTemplate,
-        messageConverter,
-        messageHeaders,
-        new UuidV4RqueueMessageIdGenerator());
+    this(messageTemplate, messageConverter, messageHeaders, new UuidV4RqueueMessageIdGenerator());
   }
 
   public RqueueMessageManagerImpl(

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/support/RqueueMessageUtils.java

@@ -72,15 +72,14 @@ public static RqueueMessage buildPeriodicMessage(
     } else {
       throw new MessageConversionException("Message payload is neither String nor byte[]");
     }
-    RqueueMessage rqueueMessage =
-        RqueueMessage.builder()
-            .id(messageIdGenerator.generate())
-            .queueName(queueName)
-            .message(strMessage)
-            .processAt(processAt)
-            .retryCount(retryCount)
-            .period(period)
-            .build();
+    RqueueMessage rqueueMessage = RqueueMessage.builder()
+        .id(messageIdGenerator.generate())
+        .queueName(queueName)
+        .message(strMessage)
+        .processAt(processAt)
+        .retryCount(retryCount)
+        .period(period)
+        .build();
     if (messageId != null) {
       rqueueMessage.setId(messageId);
     }
@@ -114,15 +113,14 @@ public static RqueueMessage buildMessage(
     } else {
       throw new MessageConversionException("Message payload is neither String nor byte[]");
     }
-    RqueueMessage rqueueMessage =
-        RqueueMessage.builder()
-            .retryCount(retryCount)
-            .queuedTime(queuedTime)
-            .id(messageIdGenerator.generate())
-            .queueName(queueName)
-            .message(strMessage)
-            .processAt(processAt)
-            .build();
+    RqueueMessage rqueueMessage = RqueueMessage.builder()
+        .retryCount(retryCount)
+        .queuedTime(queuedTime)
+        .id(messageIdGenerator.generate())
+        .queueName(queueName)
+        .message(strMessage)
+        .processAt(processAt)
+        .build();
     if (messageId != null) {
       rqueueMessage.setId(messageId);
     }
@@ -135,27 +133,13 @@ public static List<RqueueMessage> generateMessages(
       String queueName,
       int count) {
     return generateMessages(
-        messageIdGenerator,
-        converter,
-        messageIdGenerator.generate(),
-        queueName,
-        null,
-        null,
-        count);
+        messageIdGenerator, converter, messageIdGenerator.generate(), queueName, null, null, count);
   }
 
   public static RqueueMessage generateMessage(
-      RqueueMessageIdGenerator messageIdGenerator,
-      MessageConverter converter,
-      String queueName) {
+      RqueueMessageIdGenerator messageIdGenerator, MessageConverter converter, String queueName) {
     return generateMessages(
-            messageIdGenerator,
-            converter,
-            messageIdGenerator.generate(),
-            queueName,
-            null,
-            null,
-            1)
+            messageIdGenerator, converter, messageIdGenerator.generate(), queueName, null, null, 1)
         .get(0);
   }
 
@@ -185,16 +169,8 @@ public static List<RqueueMessage> generateMessages(
       int count) {
     List<RqueueMessage> messages = new ArrayList<>();
     for (int i = 0; i < count; i++) {
-      messages.add(
-          buildMessage(
-              messageIdGenerator,
-              converter,
-              queueName,
-              null,
-              object,
-              retryCount,
-              delay,
-              null));
+      messages.add(buildMessage(
+          messageIdGenerator, converter, queueName, null, object, retryCount, delay, null));
     }
     return messages;
   }

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessagePoller.java

@@ -105,8 +105,9 @@ protected boolean hasAvailableThreads(QueueDetail queueDetail, QueueThreadPool q
 
   protected void recordCapacityExhausted(QueueDetail queueDetail, QueueThreadPool queueThreadPool) {
     if (rqueueBeanProvider.getRqueueWorkerRegistry() != null) {
-      rqueueBeanProvider.getRqueueWorkerRegistry().recordQueueCapacityExhausted(
-          queueDetail, queueThreadPool);
+      rqueueBeanProvider
+          .getRqueueWorkerRegistry()
+          .recordQueueCapacityExhausted(queueDetail, queueThreadPool);
     }
   }