Add E2E tests for concurrency, consumerName override, reactive enqueue

Adds three NATS-tagged Spring Boot end-to-end integration tests plus a
shared AbstractNatsBootIT base class that lifts the Testcontainers +
@DynamicPropertySource boilerplate.

- NatsConcurrencyE2EIT: proves @RqueueListener(concurrency="3") yields
  >1 parallel handler invocations on JetStream pull subscribers.
- NatsConsumerNameOverrideE2EIT: confirms an explicit consumerName
  attribute lands on the JetStream stream as a durable consumer with
  that exact name (queried via JetStreamManagement).
- NatsReactiveEnqueueE2EIT: enqueues 5 messages via
  ReactiveRqueueMessageEnqueuer (Flux.merge) and verifies a sync
  listener receives all 5; requires rqueue.reactive.enabled=true.

Assisted-By: Claude Code

Author: sonus21

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

@@ -0,0 +1,50 @@
+/*
+ * 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 org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Common Testcontainers + dynamic-property boilerplate for NATS-backed end-to-end tests.
+ *
+ * <p>Subclasses declare their own {@code @SpringBootApplication} test config (typically excluding
+ * Redis auto-config, see {@link NatsBackendEndToEndIT} for the reference pattern) and any
+ * {@code @RqueueListener} beans they need. The container is lifecycle-managed by the
+ * {@link Testcontainers} extension and shared across all tests in a single subclass.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+abstract class AbstractNatsBootIT {
+
+  @Container
+  static final GenericContainer<?> NATS = new GenericContainer<>(
+          DockerImageName.parse("nats:2.10-alpine"))
+      .withCommand("-js")
+      .withExposedPorts(4222)
+      .waitingFor(Wait.forLogMessage(".*Server is ready.*\\n", 1));
+
+  @DynamicPropertySource
+  static void natsProps(DynamicPropertyRegistry r) {
+    r.add(
+        "rqueue.nats.connection.url",
+        () -> "nats://" + NATS.getHost() + ":" + NATS.getMappedPort(4222));
+  }
+}

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

@@ -0,0 +1,84 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+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;
+
+/**
+ * End-to-end test confirming that {@code @RqueueListener(concurrency=...)} actually runs more
+ * than one handler invocation in parallel against the NATS backend. We don't assert an exact
+ * parallelism value because JetStream prefetch + thread scheduling makes that flaky; observing
+ * any parallelism > 1 is enough proof the concurrency knob is wired through to a pull
+ * subscription with multiple poller threads.
+ */
+@SpringBootTest(
+    classes = NatsConcurrencyE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats"})
+@Tag("nats")
+class NatsConcurrencyE2EIT extends AbstractNatsBootIT {
+
+  @Autowired RqueueMessageEnqueuer enqueuer;
+
+  @Autowired ConcurrencyListener listener;
+
+  @Test
+  void parallelInvocationsAreObserved() throws Exception {
+    for (int i = 0; i < 30; i++) {
+      enqueuer.enqueue("conc-e2e", "msg-" + i);
+    }
+    assertThat(listener.latch.await(45, TimeUnit.SECONDS)).isTrue();
+    assertThat(listener.maxParallel.get())
+        .as("at least 2 concurrent invocations should have been observed")
+        .isGreaterThanOrEqualTo(2);
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class ConcurrencyListener {
+    final CountDownLatch latch = new CountDownLatch(30);
+    final AtomicInteger active = new AtomicInteger();
+    final AtomicInteger maxParallel = new AtomicInteger();
+
+    @RqueueListener(value = "conc-e2e", concurrency = "3")
+    void onMessage(String payload) throws InterruptedException {
+      int now = active.incrementAndGet();
+      maxParallel.updateAndGet(curr -> Math.max(curr, now));
+      try {
+        Thread.sleep(200L);
+      } finally {
+        active.decrementAndGet();
+        latch.countDown();
+      }
+    }
+  }
+}

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

@@ -0,0 +1,77 @@
+/*
+ * 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 io.nats.client.JetStreamManagement;
+import io.nats.client.api.ConsumerInfo;
+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 that {@code @RqueueListener(consumerName="...")} causes the JetStream durable consumer
+ * to be created with that exact name (rather than the auto-derived
+ * {@code rqueue-<queue>-<bean>#<method>} form). After a message round-trips successfully, we
+ * query the broker via {@link JetStreamManagement#getConsumerInfo} and assert the override is
+ * present.
+ */
+@SpringBootTest(
+    classes = NatsConsumerNameOverrideE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats"})
+@Tag("nats")
+class NatsConsumerNameOverrideE2EIT extends AbstractNatsBootIT {
+
+  @Autowired RqueueMessageEnqueuer enqueuer;
+
+  @Autowired CustomConsumerListener listener;
+
+  @Autowired JetStreamManagement jsm;
+
+  @Test
+  void overriddenConsumerNameIsRegisteredOnTheStream() throws Exception {
+    enqueuer.enqueue("custom-consumer", "hello");
+    assertThat(listener.latch.await(20, TimeUnit.SECONDS)).isTrue();
+
+    ConsumerInfo info = jsm.getConsumerInfo("rqueue-custom-consumer", "my-custom-consumer");
+    assertThat(info).isNotNull();
+    assertThat(info.getName()).isEqualTo("my-custom-consumer");
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class CustomConsumerListener {
+    final CountDownLatch latch = new CountDownLatch(1);
+
+    @RqueueListener(value = "custom-consumer", consumerName = "my-custom-consumer")
+    void onMessage(String payload) {
+      latch.countDown();
+    }
+  }
+}

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsReactiveEnqueueE2EIT.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.ReactiveRqueueMessageEnqueuer;
+import java.time.Duration;
+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;
+import reactor.core.publisher.Flux;
+
+/**
+ * Verifies the reactive producer path on the NATS backend: enqueueing 5 messages via
+ * {@link ReactiveRqueueMessageEnqueuer} (subscribed via {@link Flux#merge}) and confirming a
+ * synchronous {@code @RqueueListener} on the same queue receives all 5.
+ */
+@SpringBootTest(
+    classes = NatsReactiveEnqueueE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats", "rqueue.reactive.enabled=true"})
+@Tag("nats")
+class NatsReactiveEnqueueE2EIT extends AbstractNatsBootIT {
+
+  @Autowired ReactiveRqueueMessageEnqueuer reactiveEnqueuer;
+
+  @Autowired ReactiveListener listener;
+
+  @Test
+  void reactivelyEnqueuedMessagesAreReceivedByListener() throws Exception {
+    List<reactor.core.publisher.Mono<String>> publishers = new ArrayList<>();
+    for (int i = 0; i < 5; i++) {
+      publishers.add(reactiveEnqueuer.enqueue("reactive-e2e", "rx-" + i));
+    }
+    List<String> ids = Flux.merge(publishers).collectList().block(Duration.ofSeconds(15));
+    assertThat(ids).hasSize(5);
+
+    assertThat(listener.latch.await(20, TimeUnit.SECONDS)).isTrue();
+    assertThat(listener.received)
+        .containsExactlyInAnyOrder("rx-0", "rx-1", "rx-2", "rx-3", "rx-4");
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class ReactiveListener {
+    final CountDownLatch latch = new CountDownLatch(5);
+    final List<String> received = Collections.synchronizedList(new ArrayList<>());
+
+    @RqueueListener(value = "reactive-e2e")
+    void onMessage(String payload) {
+      received.add(payload);
+      latch.countDown();
+    }
+  }
+}