feat: multi-consumer support with per-consumer poller tracking and worker registry

Rename natsConsumerName -> consumerName to make the concept broker-agnostic.
Introduce pollerKey (queue##consumer) in RqueueMessageListenerContainer to
prevent the second consumer of a multi-consumer queue being skipped by the
bare-name dedup check. Wire worker registry heartbeats and capacity-exhausted
tracking through consumerTrackingKey so each consumer is reported separately.
Add NATS stream retention policy and a 14-day maxAge default; wire the message
broker onto RqueueMessageTemplateImpl at container start for non-Redis publish
paths. Extract BaseListener/JobListener in the spring-boot example.

Assisted-By: Claude Code

Author: sonus21

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

@@ -44,7 +44,17 @@
 public class RqueueConfig {
 
   @Getter
-  private static final String brokerId = UUID.randomUUID().toString();
+  private static final String brokerId = generateBrokerId();
+
+  private static String generateBrokerId() {
+    String chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
+    java.util.Random random = new java.util.Random();
+    StringBuilder sb = new StringBuilder(8);
+    for (int i = 0; i < 8; i++) {
+      sb.append(chars.charAt(random.nextInt(chars.length())));
+    }
+    return sb.toString();
+  }
 
   private static final AtomicLong counter = new AtomicLong(1);
   private final RedisConnectionFactory connectionFactory;

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

@@ -37,7 +37,15 @@
 public final class EndpointRegistry {
 
   private static final Object lock = new Object();
-  private static final Map<String, QueueDetail> queueNameToDetail = new HashMap<>();
+  // composite key: "queueName" for single-consumer, "queueName##consumerName" for multi-consumer
+  private static final Map<String, QueueDetail> registry = new HashMap<>();
+  // first registered QueueDetail per queue name — used by enqueue / management operations
+  private static final Map<String, QueueDetail> primaryByName = new HashMap<>();
+
+  private static String compositeKey(QueueDetail qd) {
+    String cn = qd.getConsumerName();
+    return (cn != null && !cn.isEmpty()) ? qd.getName() + "##" + cn : qd.getName();
+  }
 
   private EndpointRegistry() {}
 
@@ -51,7 +59,7 @@ private EndpointRegistry() {}
    * @see #get(String, String)
    */
   public static QueueDetail get(String queueName) {
-    QueueDetail queueDetail = queueNameToDetail.get(queueName);
+    QueueDetail queueDetail = primaryByName.get(queueName);
     if (queueDetail == null) {
       throw new QueueDoesNotExist(queueName);
     }
@@ -68,7 +76,7 @@ public static QueueDetail get(String queueName) {
    */
   public static QueueDetail get(String queueName, String priority) {
     QueueDetail queueDetail =
-        queueNameToDetail.get(PriorityUtils.getQueueNameForPriority(queueName, priority));
+        primaryByName.get(PriorityUtils.getQueueNameForPriority(queueName, priority));
     if (queueDetail == null) {
       throw new QueueDoesNotExist(queueName);
     }
@@ -77,24 +85,27 @@ public static QueueDetail get(String queueName, String priority) {
 
   public static void register(QueueDetail queueDetail) {
     synchronized (lock) {
-      if (queueNameToDetail.containsKey(queueDetail.getName())) {
+      String ck = compositeKey(queueDetail);
+      if (registry.containsKey(ck)) {
         throw new OverrideException(queueDetail.getName());
       }
-      queueNameToDetail.put(queueDetail.getName(), queueDetail);
+      registry.put(ck, queueDetail);
+      primaryByName.putIfAbsent(queueDetail.getName(), queueDetail);
       lock.notifyAll();
     }
   }
 
   public static void delete() {
     synchronized (lock) {
-      queueNameToDetail.clear();
+      registry.clear();
+      primaryByName.clear();
       lock.notifyAll();
     }
   }
 
   public static List<String> getActiveQueues() {
     synchronized (lock) {
-      List<String> queues = queueNameToDetail.values().stream()
+      List<String> queues = primaryByName.values().stream()
           .filter(QueueDetail::isActive)
           .map(QueueDetail::getName)
           .collect(Collectors.toList());
@@ -105,7 +116,7 @@ public static List<String> getActiveQueues() {
 
   public static List<QueueDetail> getActiveQueueDetails() {
     synchronized (lock) {
-      List<QueueDetail> queueDetails = queueNameToDetail.values().stream()
+      List<QueueDetail> queueDetails = registry.values().stream()
           .filter(QueueDetail::isActive)
           .collect(Collectors.toList());
       lock.notifyAll();
@@ -115,7 +126,7 @@ public static List<QueueDetail> getActiveQueueDetails() {
 
   public static Map<String, QueueDetail> getActiveQueueMap() {
     synchronized (lock) {
-      Map<String, QueueDetail> queueDetails = queueNameToDetail.values().stream()
+      Map<String, QueueDetail> queueDetails = primaryByName.values().stream()
           .filter(QueueDetail::isActive)
           .collect(Collectors.toMap(QueueDetail::getName, Function.identity()));
       lock.notifyAll();
@@ -126,7 +137,7 @@ public static Map<String, QueueDetail> getActiveQueueMap() {
   public static String toStr() {
     StringBuilder builder = new StringBuilder();
     synchronized (lock) {
-      List<QueueDetail> queueDetails = new ArrayList<>(queueNameToDetail.values());
+      List<QueueDetail> queueDetails = new ArrayList<>(registry.values());
       queueDetails.sort(Comparator.comparing(QueueDetail::getName));
       for (QueueDetail q : queueDetails) {
         builder.append(q.toString());
@@ -142,6 +153,6 @@ public static int getActiveQueueCount() {
   }
 
   public static int getRegisteredQueueCount() {
-    return queueNameToDetail.size();
+    return registry.size();
   }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/middleware/HandlerMiddleware.java

@@ -48,6 +48,7 @@ public void handle(Job job, Callable<Void> next) throws Exception {
         payload,
         buildMessageHeaders(
             job.getQueueDetail().getName(),
+            job.getQueueDetail().getConsumerName(),
             rqueueMessage,
             job,
             execution,

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

@@ -38,6 +38,7 @@ class DefaultRqueuePoller extends RqueueMessagePoller {
   private final QueueThreadPool queueThreadPool;
 
   DefaultRqueuePoller(
+      String pollerKey,
       QueueDetail queueDetail,
       QueueThreadPool queueThreadPool,
       RqueueBeanProvider rqueueBeanProvider,
@@ -96,6 +97,8 @@ void poll() {
       logNotAvailable();
       TimeoutUtils.sleepLog(pollingInterval, false);
     } else {
+      // Pass the pollerKey (queues[0]) so pollAndExecute's isQueueActive check uses the same
+      // key that queueRunningState was keyed with (may be "queueName##consumerName").
       super.poll(-1, queueDetail.getName(), queueDetail, queueThreadPool);
       lastNotAvailableAt = null;
     }

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

@@ -35,6 +35,9 @@ class MappingInformation implements Comparable<MappingInformation> {
   @EqualsAndHashCode.Include
   private final Set<String> queueNames;
 
+  @EqualsAndHashCode.Include
+  private final String consumerName;
+
   private final int numRetry;
   private final String deadLetterQueueName;
   private final boolean deadLetterConsumerEnabled;
@@ -46,7 +49,6 @@ class MappingInformation implements Comparable<MappingInformation> {
   private final boolean primary;
   private final int batchSize;
   private final Set<Class<? extends Throwable>> doNotRetry;
-  private final String natsConsumerName;
 
   @Override
   public String toString() {

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

@@ -84,7 +84,7 @@ public class QueueDetail extends SerializableBase {
   private final Duration natsAckWaitOverride;
   private final Integer natsMaxDeliverOverride;
   private final Duration natsDedupWindow;
-  private final String natsConsumerName;
+  private final String consumerName;
 
   public boolean isDlqSet() {
     return !StringUtils.isEmpty(deadLetterQueueName);
@@ -165,14 +165,30 @@ private QueueDetail cloneQueueDetail(
         .concurrency(concurrency)
         .priority(Collections.singletonMap(Constants.DEFAULT_PRIORITY_KEY, priority))
         .doNotRetry(doNotRetry)
-        .natsConsumerName(natsConsumerName)
+        .consumerName(consumerName)
         .build();
   }
 
   public Duration visibilityDuration() {
     return Duration.ofMillis(visibilityTimeout);
   }
 
+  /**
+   * Returns the effective JetStream consumer name for this queue. When {@link #consumerName} is
+   * explicitly set it is returned as-is. Otherwise a default is derived from the queue name:
+   * primary (non-system-generated) queues get {@code {name}-consumer-primary}; system-generated
+   * priority sub-queues get {@code {name}-consumer}. The name is sanitized so that characters
+   * outside {@code [A-Za-z0-9_-]} (e.g. the {@code ::} priority suffix separator) are replaced
+   * with {@code -}, producing a valid NATS consumer name in all cases.
+   */
+  public String resolvedConsumerName() {
+    if (consumerName != null && !consumerName.isEmpty()) {
+      return consumerName;
+    }
+    String sanitized = name.replaceAll("[^A-Za-z0-9_-]", "-");
+    return systemGenerated ? sanitized + "-consumer" : sanitized + "-consumer-primary";
+  }
+
   /**
    * Resolves the JetStream stream name. When {@link #natsStream} is null the default
    * derivation {@code "rqueue-" + queueName} is used so existing queue configs keep

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

@@ -252,9 +252,15 @@ protected void registerHandlerMethod(Object handler, Method method, MappingInfor
 
   @Override
   protected void handleMessageInternal(Message<?> message, String lookupDestination) {
-    MappingInformation mapping = this.destinationLookup.get(lookupDestination);
+    String consumerName = (String) message.getHeaders().get(RqueueMessageHeaders.CONSUMER_NAME);
+    String lookupKey = (consumerName != null && !consumerName.isEmpty())
+        ? consumerLookupKey(lookupDestination, consumerName)
+        : lookupDestination;
+    MappingInformation mapping = this.destinationLookup.get(lookupKey);
     Set<Match> matches = new HashSet<>();
-    addMatchesToCollection(mapping, message, matches);
+    if (mapping != null) {
+      addMatchesToCollection(mapping, message, matches);
+    }
     if (matches.isEmpty()) {
       handleNoMatch(this.getHandlerMethodMap().keySet(), lookupDestination, message);
       return;
@@ -327,7 +333,7 @@ private MappingInformation getMappingInformation(RqueueListener rqueueListener)
         .priority(priorityMap)
         .batchSize(batchSize)
         .doNotRetry(new HashSet<>(Arrays.asList(rqueueListener.doNotRetry())))
-        .natsConsumerName(natsConsumerName.isEmpty() ? null : natsConsumerName)
+        .consumerName(natsConsumerName.isEmpty() ? null : natsConsumerName)
         .build();
     if (mappingInformation.isValid()) {
       return mappingInformation;
@@ -523,13 +529,25 @@ private Set<String> resolveQueueNames(RqueueListener rqueueListener) {
 
   @Override
   protected Set<String> getDirectLookupDestinations(MappingInformation mapping) {
-    Set<String> destinations = new HashSet<>(mapping.getQueueNames());
+    String consumerName = mapping.getConsumerName();
+    Set<String> destinations = new HashSet<>();
     for (String queueName : mapping.getQueueNames()) {
-      destinations.addAll(PriorityUtils.getNamesFromPriority(queueName, mapping.getPriority()));
+      if (consumerName != null && !consumerName.isEmpty()) {
+        // Consumer-qualified key so two @RqueueListener methods with different consumerNames
+        // on the same queue get independent destinationLookup entries.
+        destinations.add(consumerLookupKey(queueName, consumerName));
+      } else {
+        destinations.add(queueName);
+        destinations.addAll(PriorityUtils.getNamesFromPriority(queueName, mapping.getPriority()));
+      }
     }
     return destinations;
   }
 
+  static String consumerLookupKey(String queueName, String consumerName) {
+    return queueName + "##" + consumerName;
+  }
+
   @Override
   protected String getDestination(Message<?> message) {
     return (String) message.getHeaders().get(RqueueMessageHeaders.DESTINATION);

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

@@ -56,6 +56,13 @@ public final class RqueueMessageHeaders {
    */
   public static final String EXECUTION = "execution";
 
+  /**
+   * NATS JetStream consumer name — set when a message was fetched for a specific named consumer so
+   * that {@link RqueueMessageHandler} can route it to the correct {@code @RqueueListener} method.
+   * Absent (null) for single-consumer queues and all Redis-backed queues.
+   */
+  public static final String CONSUMER_NAME = "consumerName";
+
   private static final MessageHeaders emptyMessageHeaders =
       new MessageHeaders(Collections.emptyMap());
 
@@ -71,8 +78,21 @@ public static MessageHeaders buildMessageHeaders(
       Job job,
       Execution execution,
       MessageHeaders messageHeaders) {
-    Map<String, Object> headers = new HashMap<>(9);
+    return buildMessageHeaders(destination, null, rqueueMessage, job, execution, messageHeaders);
+  }
+
+  public static MessageHeaders buildMessageHeaders(
+      String destination,
+      String consumerName,
+      RqueueMessage rqueueMessage,
+      Job job,
+      Execution execution,
+      MessageHeaders messageHeaders) {
+    Map<String, Object> headers = new HashMap<>(10);
     headers.put(DESTINATION, destination);
+    if (consumerName != null && !consumerName.isEmpty()) {
+      headers.put(CONSUMER_NAME, consumerName);
+    }
     headers.put(ID, rqueueMessage.getId());
     headers.put(MESSAGE, rqueueMessage);
     if (job != null) {