Add E2E tests for priority queues and multi-listener fan-out on NATS

- NatsPriorityQueuesE2EIT: a single @RqueueListener with
  priority="high=10,low=1" consumes 5 high + 5 low messages enqueued
  via RqueueMessageEnqueuer.enqueueWithPriority and we assert per-
  priority counts come out exact.
- NatsMultipleListenersOnSameQueueE2EIT: documents the desired
  fan-out semantics across two @RqueueListener methods on the same
  queue, but is @disabled because the default WorkQueue stream
  retention only allows a single filter-overlapping consumer; enable
  once the broker supports Limits/Interest retention per queue.

Assisted-By: Claude Code

Author: sonus21

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsMultipleListenersOnSameQueueE2EIT.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2024-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.spring.boot.integration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.sonus21.rqueue.annotation.RqueueListener;
+import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisReactiveAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.stereotype.Component;
+
+/**
+ * Two {@code @RqueueListener} methods on the same queue should each receive every message
+ * published to it (independent durable consumers, fan-out semantics). Currently disabled because
+ * the default JetStream stream retention is {@code WorkQueue}, which permits only one
+ * filter-overlapping consumer at a time and deletes a message after the first ack — so true
+ * fan-out is not possible with the v1 broker defaults. To enable this test, the broker would
+ * need to switch to {@code Limits} or {@code Interest} retention for queues with multiple
+ * listeners.
+ */
+@SpringBootTest(
+    classes = NatsMultipleListenersOnSameQueueE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats"})
+@Tag("nats")
+@Disabled(
+    "Default JetStream retention=WorkQueue prevents true fan-out across multiple consumers; "
+        + "enable once retention is configurable per queue or defaulted to Limits/Interest.")
+class NatsMultipleListenersOnSameQueueE2EIT extends AbstractNatsBootIT {
+
+  @Autowired RqueueMessageEnqueuer enqueuer;
+
+  @Autowired ListenerOne one;
+
+  @Autowired ListenerTwo two;
+
+  @Test
+  void bothListenersReceiveAllMessages() throws Exception {
+    for (int i = 0; i < 5; i++) {
+      enqueuer.enqueue("multi", "fan-" + i);
+    }
+    assertThat(one.latch.await(20, TimeUnit.SECONDS)).isTrue();
+    assertThat(two.latch.await(20, TimeUnit.SECONDS)).isTrue();
+    assertThat(one.received).hasSize(5);
+    assertThat(two.received).hasSize(5);
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class ListenerOne {
+    final CountDownLatch latch = new CountDownLatch(5);
+    final List<String> received = Collections.synchronizedList(new ArrayList<>());
+
+    @RqueueListener(value = "multi")
+    void onMessage(String payload) {
+      received.add(payload);
+      latch.countDown();
+    }
+  }
+
+  @Component
+  static class ListenerTwo {
+    final CountDownLatch latch = new CountDownLatch(5);
+    final List<String> received = Collections.synchronizedList(new ArrayList<>());
+
+    @RqueueListener(value = "multi")
+    void onMessage(String payload) {
+      received.add(payload);
+      latch.countDown();
+    }
+  }
+}

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsPriorityQueuesE2EIT.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2024-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.spring.boot.integration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.sonus21.rqueue.annotation.RqueueListener;
+import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisReactiveAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.stereotype.Component;
+
+/**
+ * Verifies queue-level priority on the NATS backend: a single listener with
+ * {@code priority="high=10,low=1"} consumes from two internal sub-queues
+ * ({@code pq_high} and {@code pq_low}) and the producer sends to each via
+ * {@link RqueueMessageEnqueuer#enqueueWithPriority}. We assert all 10 messages are
+ * received and that 5 messages with payload prefix "high-" and 5 with "low-" arrive.
+ */
+@SpringBootTest(
+    classes = NatsPriorityQueuesE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats"})
+@Tag("nats")
+class NatsPriorityQueuesE2EIT extends AbstractNatsBootIT {
+
+  @Autowired RqueueMessageEnqueuer enqueuer;
+
+  @Autowired PriorityListener listener;
+
+  @Test
+  void messagesEnqueuedAtBothPrioritiesAreReceived() throws Exception {
+    for (int i = 0; i < 5; i++) {
+      enqueuer.enqueueWithPriority("pq", "high", "high-" + i);
+      enqueuer.enqueueWithPriority("pq", "low", "low-" + i);
+    }
+    assertThat(listener.latch.await(30, TimeUnit.SECONDS)).isTrue();
+
+    long highCount = listener.received.stream().filter(s -> s.startsWith("high-")).count();
+    long lowCount = listener.received.stream().filter(s -> s.startsWith("low-")).count();
+    assertThat(highCount).isEqualTo(5);
+    assertThat(lowCount).isEqualTo(5);
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class PriorityListener {
+    final CountDownLatch latch = new CountDownLatch(10);
+    final List<String> received = Collections.synchronizedList(new ArrayList<>());
+
+    @RqueueListener(value = "pq", priority = "high=10,low=1")
+    void onMessage(String payload) {
+      received.add(payload);
+      latch.countDown();
+    }
+  }
+}