Merge branch 'nats-scheduling-fix' of github.com:/sonus21/rqueue into nats-scheduling-fix

Author: sonus21

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -253,7 +253,8 @@ public void ensureStream(
           // enqueueWithDelay after the stream was originally created by a plain enqueue call).
           java.util.List<String> existingSubjects = existing.getConfiguration().getSubjects();
           java.util.Set<String> existingSet = existingSubjects != null
-              ? new java.util.HashSet<>(existingSubjects) : new java.util.HashSet<>();
+              ? new java.util.HashSet<>(existingSubjects)
+              : new java.util.HashSet<>();
           boolean needsSubjectUpdate = subjects.stream().anyMatch(s -> !existingSet.contains(s));
           boolean needsFlagUpdate =
               enableSchedules && !existing.getConfiguration().getAllowMsgSchedules();
@@ -262,19 +263,22 @@ public void ensureStream(
             // Merge: keep all existing subjects and append new ones (never remove).
             java.util.LinkedHashSet<String> merged = new java.util.LinkedHashSet<>(existingSet);
             merged.addAll(subjects);
-            StreamConfiguration.Builder upd =
-                StreamConfiguration.builder(existing.getConfiguration())
-                    .subjects(new java.util.ArrayList<>(merged));
+            StreamConfiguration.Builder upd = StreamConfiguration.builder(
+                    existing.getConfiguration())
+                .subjects(new java.util.ArrayList<>(merged));
             if (needsFlagUpdate) {
               upd.allowMessageSchedules(true);
             }
             jsm.updateStream(upd.build());
             if (needsFlagUpdate) {
-              log.log(Level.INFO,
-                  "Stream ''{0}'' updated to enable message scheduling (ADR-51).", streamName);
+              log.log(
+                  Level.INFO,
+                  "Stream ''{0}'' updated to enable message scheduling (ADR-51).",
+                  streamName);
             }
             if (needsSubjectUpdate) {
-              log.log(Level.INFO,
+              log.log(
+                  Level.INFO,
                   "Stream ''{0}'' updated with additional subjects: {1}.",
                   new Object[] {streamName, subjects});
             }

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/JetStreamMessageBrokerSchedulingIT.java

@@ -171,10 +171,8 @@ void enqueueWithDelay_afterPlainEnqueue_streamUpgradedAndMessageHeld() throws Ex
               + ")");
 
       // Step 1 — plain enqueue: creates stream with only the work subject, no sched flag
-      RqueueMessage immediate = RqueueMessage.builder()
-          .id("imm-1")
-          .message("immediate")
-          .build();
+      RqueueMessage immediate =
+          RqueueMessage.builder().id("imm-1").message("immediate").build();
       broker.enqueue(q, immediate);
 
       // Step 2 — delayed enqueue: must upgrade the stream (add sched wildcard + flag)

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/internal/NatsProvisionerTest.java

@@ -323,16 +323,12 @@ void ensureStream_firstEnqueueThenDelay_addsSchedSubjectAndFlag()
 
     // enqueueWithDelay() calls ensureStream with the sched wildcard and allowSchedules=true
     provisioner.ensureStream(
-        "rqueue-js-orders",
-        Arrays.asList(workSubject, schedPattern),
-        QueueType.QUEUE,
-        null,
-        true);
+        "rqueue-js-orders", Arrays.asList(workSubject, schedPattern), QueueType.QUEUE, null, true);
 
     // Must call updateStream (never addStream — stream already exists)
     verify(jsm, never()).addStream(any());
-    verify(jsm, times(1)).updateStream(argThat(cfg ->
-        cfg.getAllowMsgSchedules()
+    verify(jsm, times(1))
+        .updateStream(argThat(cfg -> cfg.getAllowMsgSchedules()
             && cfg.getSubjects().contains(workSubject)
             && cfg.getSubjects().contains(schedPattern)));
   }

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

@@ -78,6 +78,7 @@ static void wipeOwnedStreams() {
    * cannot race with the scheduler firing.
    */
   static final Duration ETD_DELAY = Duration.ofSeconds(10);
+
   static final Duration ETD_TOTAL_WAIT = Duration.ofSeconds(20);
 
   @Autowired
@@ -204,7 +205,7 @@ static class EtdListener {
     void onMessage(String payload) {
       received.add(payload);
       immediateLatch.countDown(); // first call unblocks; subsequent calls are no-ops
-      allLatch.countDown();       // each call decrements; reaches 0 when both arrive
+      allLatch.countDown(); // each call decrements; reaches 0 when both arrive
     }
   }