add worker heart beat

Author: sonus21

rqueue-core/src/main/java/com/github/sonus21/rqueue/common/RqueueRedisTemplate.java

@@ -66,6 +66,10 @@ public Long rpush(String listName, V val) {
     return redisTemplate.opsForList().rightPush(listName, val);
   }
 
+  public Long lpush(String listName, V val) {
+    return redisTemplate.opsForList().leftPush(listName, val);
+  }
+
   public Long addToSet(String setName, V... values) {
     return redisTemplate.opsForSet().add(setName, values);
   }
@@ -98,6 +102,23 @@ public void set(String key, V val, Duration duration) {
     redisTemplate.opsForValue().set(key, val, duration.toMillis(), TimeUnit.MILLISECONDS);
   }
 
+  public void putHashValue(String key, String hashKey, V val) {
+    redisTemplate.opsForHash().put(key, hashKey, val);
+  }
+
+  @SuppressWarnings("unchecked")
+  public Map<String, V> getHashEntries(String key) {
+    return (Map<String, V>) (Map<?, ?>) redisTemplate.opsForHash().entries(key);
+  }
+
+  public Long deleteHashValues(String key, String... hashKeys) {
+    return redisTemplate.opsForHash().delete(key, (Object[]) hashKeys);
+  }
+
+  public Boolean expire(String key, Duration duration) {
+    return redisTemplate.expire(key, duration.toMillis(), TimeUnit.MILLISECONDS);
+  }
+
   public Boolean setIfAbsent(String lockKey, V val, Duration duration) {
     boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, val);
     if (result) {

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

@@ -43,6 +43,7 @@
 @Configuration
 public class RqueueConfig {
 
+  @Getter
   private static final String brokerId = UUID.randomUUID().toString();
   private static final AtomicLong counter = new AtomicLong(1);
   private final RedisConnectionFactory connectionFactory;
@@ -139,9 +140,26 @@ public class RqueueConfig {
   @Value("${rqueue.completed.job.cleanup.interval:30000}")
   private long completedJobCleanupIntervalInMs;
 
-  public static String getBrokerId() {
-    return brokerId;
-  }
+  @Value("${rqueue.worker.registry.enabled:true}")
+  private boolean workerRegistryEnabled;
+
+  @Value("${rqueue.worker.registry.worker.ttl:300}")
+  private long workerRegistryWorkerTtlInSeconds;
+
+  @Value("${rqueue.worker.registry.worker.heartbeat.interval:60}")
+  private long workerRegistryWorkerHeartbeatIntervalInSeconds;
+
+  @Value("${rqueue.worker.registry.queue.ttl:3600}")
+  private long workerRegistryQueueTtlInSeconds;
+
+  @Value("${rqueue.worker.registry.queue.heartbeat.interval:15}")
+  private long workerRegistryQueueHeartbeatIntervalInSeconds;
+
+  @Value("${rqueue.worker.registry.key.prefix:worker::}")
+  private String workerRegistryKeyPrefix;
+
+  @Value("${rqueue.worker.registry.queue.key.prefix:q-pollers::}")
+  private String workerRegistryQueueKeyPrefix;
 
   public boolean messageInTerminalStateShouldBeStored() {
     return getMessageDurabilityInTerminalStateInSecond() > 0;
@@ -294,6 +312,14 @@ public String getJobsKey(String messageId) {
     return prefix + jobsCollectionNamePrefix + messageId;
   }
 
+  public String getWorkerRegistryKey(String workerId) {
+    return prefix + workerRegistryKeyPrefix + workerId;
+  }
+
+  public String getWorkerRegistryQueueKey(String queueName) {
+    return prefix + workerRegistryQueueKeyPrefix + getTaggedName(queueName);
+  }
+
   public String getDelDataName(String queueName) {
     return prefix
         + delPrefix
@@ -307,6 +333,22 @@ public Duration getJobDurabilityInTerminalState() {
     return Duration.ofSeconds(jobDurabilityInTerminalStateInSecond);
   }
 
+  public Duration getWorkerRegistryWorkerTtl() {
+    return Duration.ofSeconds(workerRegistryWorkerTtlInSeconds);
+  }
+
+  public Duration getWorkerRegistryWorkerHeartbeatInterval() {
+    return Duration.ofSeconds(workerRegistryWorkerHeartbeatIntervalInSeconds);
+  }
+
+  public Duration getWorkerRegistryQueueTtl() {
+    return Duration.ofSeconds(workerRegistryQueueTtlInSeconds);
+  }
+
+  public Duration getWorkerRegistryQueueHeartbeatInterval() {
+    return Duration.ofSeconds(workerRegistryQueueHeartbeatIntervalInSeconds);
+  }
+
   public String getLibVersion() {
     if (StringUtils.isEmpty(version)) {
       ClassPathResource resource =

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

@@ -25,15 +25,20 @@
 import com.github.sonus21.rqueue.core.ProcessingQueueMessageScheduler;
 import com.github.sonus21.rqueue.core.RqueueBeanProvider;
 import com.github.sonus21.rqueue.core.RqueueInternalPubSubChannel;
+import com.github.sonus21.rqueue.core.RqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.RqueueRedisListenerContainerFactory;
 import com.github.sonus21.rqueue.core.ScheduledQueueMessageScheduler;
 import com.github.sonus21.rqueue.core.impl.RqueueMessageTemplateImpl;
+import com.github.sonus21.rqueue.core.impl.UuidV4RqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.dao.RqueueStringDao;
 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;
@@ -151,6 +156,12 @@ public RqueueWebConfig rqueueWebConfig() {
     return new RqueueWebConfig();
   }
 
+  @Bean
+  @Conditional(MissingRqueueMessageIdGenerator.class)
+  public RqueueMessageIdGenerator rqueueMessageIdGenerator() {
+    return new UuidV4RqueueMessageIdGenerator();
+  }
+
   @Bean
   public RqueueSchedulerConfig rqueueSchedulerConfig() {
     return new RqueueSchedulerConfig();
@@ -215,6 +226,11 @@ public RqueueStringDao rqueueStringDao(RqueueConfig rqueueConfig) {
     return new RqueueStringDaoImpl(rqueueConfig);
   }
 
+  @Bean
+  public RqueueWorkerRegistry rqueueWorkerRegistry(RqueueConfig rqueueConfig) {
+    return new RqueueWorkerRegistryImpl(rqueueConfig);
+  }
+
   @Bean
   public RqueueLockManager rqueueLockManager(RqueueStringDao rqueueStringDao) {
     return new RqueueLockManagerImpl(rqueueStringDao);

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

@@ -48,6 +48,12 @@ public class RqueueWebConfig {
   @Value("${rqueue.web.max.message.move.count:1000}")
   private int maxMessageMoveCount;
 
+  @Value("${rqueue.web.queue.page.size:12}")
+  private int queuePageSize;
+
+  @Value("${rqueue.web.worker.page.size:10}")
+  private int workerPageSize;
+
   /**
    * Whether queue stats should be collected or not. When this flag is disabled, metric data won't
    * be available in the dashboard.

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

@@ -24,6 +24,7 @@
 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 lombok.Getter;
 import lombok.Setter;
@@ -55,6 +56,9 @@ public class RqueueBeanProvider {
   @Autowired(required = false)
   private RqueueMetricsCounter rqueueMetricsCounter;
 
+  @Autowired(required = false)
+  private RqueueWorkerRegistry rqueueWorkerRegistry;
+
   @Autowired
   private RqueueMessageHandler rqueueMessageHandler;
 

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

@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2026 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.core;
+
+@FunctionalInterface
+public interface RqueueMessageIdGenerator {
+
+  String generate();
+}

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

@@ -49,6 +49,8 @@ void moveMessageWithDelay(
 
   Long addMessage(String queueName, RqueueMessage rqueueMessage);
 
+  Long addMessageAtFront(String queueName, RqueueMessage rqueueMessage);
+
   Boolean addToZset(String zsetName, RqueueMessage rqueueMessage, long score);
 
   List<RqueueMessage> getAllMessages(

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

@@ -26,6 +26,7 @@
 import com.github.sonus21.rqueue.config.RqueueConfig;
 import com.github.sonus21.rqueue.core.EndpointRegistry;
 import com.github.sonus21.rqueue.core.RqueueMessage;
+import com.github.sonus21.rqueue.core.RqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.impl.MessageSweeper.MessageDeleteRequest;
 import com.github.sonus21.rqueue.dao.RqueueStringDao;
@@ -51,25 +52,23 @@ abstract class BaseMessageSender {
   protected final MessageHeaders messageHeaders;
   protected final MessageConverter messageConverter;
   protected final RqueueMessageTemplate messageTemplate;
-
-  @Autowired
-  protected RqueueStringDao rqueueStringDao;
-
-  @Autowired
-  protected RqueueConfig rqueueConfig;
-
-  @Autowired
-  protected RqueueMessageMetadataService rqueueMessageMetadataService;
+  protected final RqueueMessageIdGenerator messageIdGenerator;
+  @Autowired protected RqueueStringDao rqueueStringDao;
+  @Autowired protected RqueueConfig rqueueConfig;
+  @Autowired protected RqueueMessageMetadataService rqueueMessageMetadataService;
 
   BaseMessageSender(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
-      MessageHeaders messageHeaders) {
+      MessageHeaders messageHeaders,
+      RqueueMessageIdGenerator messageIdGenerator) {
     notNull(messageTemplate, "messageTemplate cannot be null");
     notNull(messageConverter, "messageConverter cannot be null");
+    notNull(messageIdGenerator, "messageIdGenerator cannot be null");
     this.messageTemplate = messageTemplate;
     this.messageConverter = messageConverter;
     this.messageHeaders = messageHeaders;
+    this.messageIdGenerator = messageIdGenerator;
   }
 
   protected Object storeMessageMetadata(
@@ -119,14 +118,16 @@ protected String pushMessage(
       Long delayInMilliSecs,
       boolean isUnique) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage = buildMessage(
-        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);
@@ -146,14 +147,16 @@ protected String pushMessage(
   protected String pushPeriodicMessage(
       String queueName, String messageId, Object message, long periodInMilliSeconds) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage = buildPeriodicMessage(
-        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/MessageSweeper.java

@@ -42,9 +42,7 @@
 
 @Slf4j
 public class MessageSweeper {
-
   private static MessageSweeper messageSweeper;
-
   private final ExecutorService executorService;
   private final RqueueMessageTemplate messageTemplate;
   private final RqueueMessageMetadataService rqueueMessageMetadataService;

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

@@ -21,6 +21,7 @@
 import com.github.sonus21.rqueue.core.EndpointRegistry;
 import com.github.sonus21.rqueue.core.ReactiveRqueueMessageEnqueuer;
 import com.github.sonus21.rqueue.core.RqueueMessage;
+import com.github.sonus21.rqueue.core.RqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.support.RqueueMessageUtils;
 import com.github.sonus21.rqueue.exception.DuplicateMessageException;
@@ -41,7 +42,19 @@ public ReactiveRqueueMessageEnqueuerImpl(
       RqueueMessageTemplate messageTemplate,
       MessageConverter messageConverter,
       MessageHeaders messageHeaders) {
-    super(messageTemplate, messageConverter, messageHeaders);
+    this(
+        messageTemplate,
+        messageConverter,
+        messageHeaders,
+        new UuidV4RqueueMessageIdGenerator());
+  }
+
+  public ReactiveRqueueMessageEnqueuerImpl(
+      RqueueMessageTemplate messageTemplate,
+      MessageConverter messageConverter,
+      MessageHeaders messageHeaders,
+      RqueueMessageIdGenerator messageIdGenerator) {
+    super(messageTemplate, messageConverter, messageHeaders, messageIdGenerator);
   }
 
   @SuppressWarnings("unchecked")
@@ -55,14 +68,16 @@ private <T> Mono<T> pushReactiveMessage(
       boolean isUnique,
       Function<RqueueMessage, Mono<T>> monoConverter) {
     QueueDetail queueDetail = EndpointRegistry.get(queueName);
-    RqueueMessage rqueueMessage = builder.build(
-        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);
@@ -280,6 +295,7 @@ private interface MonoConverter<T> {
   @FunctionalInterface
   private interface MessageBuilder {
     RqueueMessage build(
+        RqueueMessageIdGenerator messageIdGenerator,
         MessageConverter converter,
         String queueName,
         String messageId,