Delete BrokerMessagePoller; route NATS through DefaultRqueuePoller

Replace the separate BrokerMessagePoller/ConsumerNameResolver path with
the existing DefaultRqueuePoller + QueueThreadPool + RqueueExecutor
pipeline used by the Redis backend.

Key changes:
- Delete BrokerMessagePoller, ConsumerNameResolver and their tests.
- Add backend-neutral methods to MessageBroker SPI: parkForRetry,
  moveToDlq, scheduleNext, getVisibilityTimeoutScore,
  extendVisibilityTimeout. RedisMessageBroker overrides each with the
  existing Redis template calls.
- Refactor PostProcessingHandler, JobImpl, RqueueExecutor and
  RqueueMessagePoller to use MessageBroker instead of the Redis-specific
  RqueueMessageTemplate.
- RqueueMessageListenerContainer.initialize() resolves an effective
  broker (injected or a RedisMessageBroker wrapper) and removes the
  usesPrimaryHandlerDispatch==false branch that wired the old path.
- JetStreamMessageBroker: set usesPrimaryHandlerDispatch=true so NATS
  queues flow through the standard poller; add resolveConsumerName
  fallback so a null consumerName from RqueueMessagePoller gets a stable
  "rqueue-<queue>" default.
- Update all affected tests to mock MessageBroker instead of
  RqueueMessageTemplate.

Assisted-By: Claude Code

Author: sonus21

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

@@ -19,6 +19,7 @@
 import com.github.sonus21.rqueue.common.RqueueLockManager;
 import com.github.sonus21.rqueue.config.RqueueConfig;
 import com.github.sonus21.rqueue.config.RqueueWebConfig;
+import com.github.sonus21.rqueue.core.spi.MessageBroker;
 import com.github.sonus21.rqueue.core.support.MessageProcessor;
 import com.github.sonus21.rqueue.dao.RqueueJobDao;
 import com.github.sonus21.rqueue.dao.RqueueSystemConfigDao;
@@ -69,4 +70,7 @@ public class RqueueBeanProvider {
 
   @Autowired
   private RqueueConfig rqueueConfig;
+
+  /** Set by the container during initialization; never null after {@code afterPropertiesSet}. */
+  private MessageBroker messageBroker;
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/spi/MessageBroker.java

@@ -91,6 +91,60 @@ default List<RqueueMessage> pop(
 
   List<RqueueMessage> peek(QueueDetail q, long offset, long count);
 
+  /**
+   * Remove {@code old} from the processing store and re-enqueue {@code updated} for retry.
+   * {@code delayMs <= 0} means immediate; {@code delayMs > 0} means schedule after that delay.
+   * Backends without a processing store (e.g. NATS) default to a plain nack.
+   */
+  default void parkForRetry(QueueDetail q, RqueueMessage old, RqueueMessage updated, long delayMs) {
+    nack(q, updated, delayMs);
+  }
+
+  /**
+   * Remove {@code old} from the processing store and enqueue {@code updated} to {@code targetQueue}.
+   * {@code delayMs <= 0} means immediate (list push); {@code delayMs > 0} means schedule (sorted-set).
+   * Backends without a processing store default to a plain enqueue to the DLQ.
+   */
+  default void moveToDlq(
+      QueueDetail source,
+      String targetQueue,
+      RqueueMessage old,
+      RqueueMessage updated,
+      long delayMs) {
+    if (delayMs > 0) {
+      enqueueWithDelay(source, updated, delayMs);
+    } else {
+      enqueue(source, updated);
+    }
+  }
+
+  /**
+   * Schedule the next execution of a periodic message.
+   * {@code messageKey} is the deduplication key; {@code expirySeconds} is the TTL for that key.
+   * Backends that don't support server-side scheduling default to a delayed enqueue.
+   */
+  default void scheduleNext(
+      QueueDetail q, String messageKey, RqueueMessage message, long expirySeconds) {
+    long delayMs = Math.max(0, message.getProcessAt() - System.currentTimeMillis());
+    enqueueWithDelay(q, message, delayMs);
+  }
+
+  /**
+   * Returns the score (epoch-ms deadline) of {@code m} in the processing store, or {@code null}
+   * if the backend does not track per-message visibility (e.g. NATS uses consumer-level AckWait).
+   */
+  default Long getVisibilityTimeoutScore(QueueDetail q, RqueueMessage m) {
+    return null;
+  }
+
+  /**
+   * Adds {@code deltaMs} to the visibility timeout of {@code m} in the processing store.
+   * Returns {@code false} if the backend does not support per-message lease extension.
+   */
+  default boolean extendVisibilityTimeout(QueueDetail q, RqueueMessage m, long deltaMs) {
+    return false;
+  }
+
   long size(QueueDetail q);
 
   AutoCloseable subscribe(String channel, Consumer<String> handler);

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/spi/redis/RedisMessageBroker.java

@@ -22,6 +22,7 @@
 import com.github.sonus21.rqueue.core.spi.MessageBroker;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.models.MessageMoveResult;
+import com.github.sonus21.rqueue.utils.RedisUtils;
 import java.time.Duration;
 import java.util.List;
 import java.util.function.Consumer;
@@ -146,6 +147,56 @@ public void publish(String channel, String payload) {
     template.getTemplate().convertAndSend(channel, payload);
   }
 
+  @Override
+  public void parkForRetry(QueueDetail q, RqueueMessage old, RqueueMessage updated, long delayMs) {
+    if (delayMs <= 0) {
+      template.moveMessage(q.getProcessingQueueName(), q.getQueueName(), old, updated);
+    } else {
+      template.moveMessageWithDelay(
+          q.getProcessingQueueName(), q.getScheduledQueueName(), old, updated, delayMs);
+    }
+  }
+
+  @Override
+  public void moveToDlq(
+      QueueDetail source,
+      String targetQueue,
+      RqueueMessage old,
+      RqueueMessage updated,
+      long delayMs) {
+    RedisUtils.executePipeLine(
+        template.getTemplate(),
+        (connection, keySerializer, valueSerializer) -> {
+          byte[] updatedBytes = valueSerializer.serialize(updated);
+          byte[] oldBytes = valueSerializer.serialize(old);
+          byte[] processingQueueBytes =
+              keySerializer.serialize(source.getProcessingQueueName());
+          byte[] targetQueueBytes = keySerializer.serialize(targetQueue);
+          if (delayMs > 0) {
+            connection.zAdd(targetQueueBytes, delayMs, updatedBytes);
+          } else {
+            connection.lPush(targetQueueBytes, updatedBytes);
+          }
+          connection.zRem(processingQueueBytes, oldBytes);
+        });
+  }
+
+  @Override
+  public void scheduleNext(
+      QueueDetail q, String messageKey, RqueueMessage message, long expirySeconds) {
+    template.scheduleMessage(q.getScheduledQueueName(), messageKey, message, expirySeconds);
+  }
+
+  @Override
+  public Long getVisibilityTimeoutScore(QueueDetail q, RqueueMessage m) {
+    return template.getScore(q.getProcessingQueueName(), m);
+  }
+
+  @Override
+  public boolean extendVisibilityTimeout(QueueDetail q, RqueueMessage m, long deltaMs) {
+    return template.addScore(q.getProcessingQueueName(), m, deltaMs);
+  }
+
   @Override
   public Capabilities capabilities() {
     return Capabilities.REDIS_DEFAULTS;