add fixes after review

(cherry picked from commit 21d6d86)

Author: igorjava2025

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

@@ -18,6 +18,7 @@
 
 import com.github.sonus21.rqueue.models.MessageMoveResult;
 import java.util.List;
+import java.util.Optional;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import reactor.core.publisher.Flux;
@@ -92,4 +93,10 @@ Long scheduleMessage(
 
   Flux<Long> addReactiveMessageWithDelay(
       String scheduledQueueName, String scheduledQueueChannelName, RqueueMessage rqueueMessage);
+
+  Optional<RqueueMessage> findFirstElementFromList(String name);
+
+  Optional<RqueueMessage> findFirstElementFromZset(String name);
+
+  Optional<TypedTuple<RqueueMessage>> findFirstElementFromZsetWithScore(String name);
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/RqueueMessageTemplateImpl.java

@@ -30,6 +30,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
@@ -343,4 +344,19 @@ public RedisTemplate<String, RqueueMessage> getTemplate() {
   public Long removeElementFromZset(String zsetName, RqueueMessage rqueueMessage) {
     return super.removeFromZset(zsetName, rqueueMessage);
   }
+
+  @Override
+  public Optional<RqueueMessage> findFirstElementFromList(String name) {
+    return readFromList(name, 0, 0).stream().findFirst();
+  }
+
+  @Override
+  public Optional<RqueueMessage> findFirstElementFromZset(String name) {
+    return readFromZset(name, 0, 0).stream().findFirst();
+  }
+
+  @Override
+  public Optional<TypedTuple<RqueueMessage>> findFirstElementFromZsetWithScore(String name) {
+    return readFromZsetWithScore(name, 0, 0).stream().findFirst();
+  }
 }

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

@@ -151,23 +151,37 @@ boolean existMessagesInCurrentQueueOrHigherPriorityQueue(String currentQueue, Li
   }
 
   protected boolean existAvailableMessagesForPoll(QueueDetail queueDetail) {
-    List<?> readyMessages = rqueueBeanProvider
-        .getRqueueMessageTemplate()
-        .readFromList(queueDetail.getQueueName(), 0, 0);
-
-    if (readyMessages != null && !readyMessages.isEmpty()) {
-      log(Level.TRACE, "readyMessages exists for queue '{}', existAvailableMessagesForPoll = true.", null, queueDetail.getName());
+    boolean readyMessagesExists =
+        rqueueBeanProvider
+            .getRqueueMessageTemplate()
+            .findFirstElementFromList(queueDetail.getQueueName())
+            .isPresent();
+    if (readyMessagesExists) {
+      log(
+          Level.TRACE,
+          "readyMessages exists for queue '{}', existAvailableMessagesForPoll = true.",
+          null,
+          queueDetail.getName());
       return true;
     }
 
     // Only check delayed messages with score <= current time
     long currentTime = System.currentTimeMillis();
-    List<?> delayedMessages = rqueueBeanProvider
-        .getRqueueMessageTemplate()
-        .readFromZsetWithScore(queueDetail.getScheduledQueueName(), 0, currentTime);
-
-    if (delayedMessages != null && !delayedMessages.isEmpty()) {
-      log(Level.TRACE, "delayedMessages exists for queue '{}', existAvailableMessagesForPoll = true.", null, queueDetail.getName());
+    boolean delayedMessagesExists =
+        rqueueBeanProvider
+            .getRqueueMessageTemplate()
+            .findFirstElementFromZsetWithScore(queueDetail.getScheduledQueueName())
+            .filter(element -> element.getScore() <= currentTime)
+            .isPresent();
+
+    if (delayedMessagesExists) {
+      log(
+          Level.TRACE,
+          "delayedMessages exists for scheduled queue '{}' and currentTime '{}',"
+              + " existAvailableMessagesForPoll = true.",
+          null,
+          queueDetail.getScheduledQueueName(),
+          currentTime);
       return true;
     }
 

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

@@ -11,6 +11,7 @@ public Long getAfterPollSleepInterval() {
   }
 
   public void setAfterPollSleepInterval(Long afterPollSleepInterval) {
+    validateTimeInterval(afterPollSleepInterval);
     this.afterPollSleepInterval = afterPollSleepInterval;
   }
 
@@ -19,6 +20,14 @@ public Long getSemaphoreWaitTime() {
   }
 
   public void setSemaphoreWaitTime(Long semaphoreWaitTime) {
+    validateTimeInterval(semaphoreWaitTime);
     this.semaphoreWaitTime = semaphoreWaitTime;
   }
+
+  private void validateTimeInterval(Long value) {
+    if (value == null || value > 0) {
+      return;
+    }
+    throw new IllegalArgumentException("Value must be positive: " + value);
+  }
 }

rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/HardStrictPriorityPollerPropertiesTest.java

@@ -0,0 +1,49 @@
+package com.github.sonus21.rqueue.listener;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.stream.LongStream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.NullSource;
+
+class HardStrictPriorityPollerPropertiesTest {
+
+  private final HardStrictPriorityPollerProperties properties =
+      new HardStrictPriorityPollerProperties();
+
+  static LongStream invalidValues() {
+    return LongStream.of(0L, -1L, -100L, Long.MIN_VALUE);
+  }
+
+  static LongStream validValues() {
+    return LongStream.of(1L, Long.MAX_VALUE);
+  }
+
+  @ParameterizedTest
+  @MethodSource("invalidValues")
+  void setAfterPollSleepIntervalInvalidValues(Long value) {
+    assertThrows(IllegalArgumentException.class, () -> properties.setAfterPollSleepInterval(value));
+  }
+
+  @ParameterizedTest
+  @NullSource
+  @MethodSource("validValues")
+  void setAfterPollSleepIntervalValidValues(Long value) {
+    assertDoesNotThrow(() -> properties.setAfterPollSleepInterval(value));
+  }
+
+  @ParameterizedTest
+  @MethodSource("invalidValues")
+  void setSemaphoreWaitTimeInvalidValues(Long value) {
+    assertThrows(IllegalArgumentException.class, () -> properties.setSemaphoreWaitTime(value));
+  }
+
+  @ParameterizedTest
+  @NullSource
+  @MethodSource("validValues")
+  void setSemaphoreWaitTimeValidValues(Long value) {
+    assertDoesNotThrow(() -> properties.setSemaphoreWaitTime(value));
+  }
+}