Merge pull request #58 from agorapulse/fix/composable-annotations

composable annotations
agorapulse · Dec 11, 2024 · bcf62a3 · bcf62a3
2 parents d817540 + 16e38c5
commit bcf62a3
Show file tree

Hide file tree

Showing 5 changed files with 231 additions and 87 deletions.
diff --git a/libs/micronaut-worker/src/main/java/com/agorapulse/worker/JobConfiguration.java b/libs/micronaut-worker/src/main/java/com/agorapulse/worker/JobConfiguration.java
@@ -74,6 +74,8 @@ interface QueueConfiguration {
     interface ConsumerQueueConfiguration extends QueueConfiguration {
 
         int DEFAULT_MAX_MESSAGES = 10;
+        Duration DEFAULT_WAITING_TIME = Duration.ofSeconds(20);
+        String DEFAULT_WAITING_TIME_STRING = "20s";
 
         /**
          * @return the number of messages which are fetched from the queue in a single poll, defaults to {@link #DEFAULT_MAX_MESSAGES} when set to <code>0</code>

diff --git a/...aut-worker/src/main/java/com/agorapulse/worker/configuration/DefaultJobConfiguration.java b/...aut-worker/src/main/java/com/agorapulse/worker/configuration/DefaultJobConfiguration.java
@@ -119,7 +119,11 @@ public void mergeWith(ConsumerQueueConfiguration overrides) {
                 this.maxMessages = overrides.getMaxMessages();
             }
 
-            if (overrides.getWaitingTime() != null && !overrides.getWaitingTime().isZero() && overrides.getWaitingTime() != this.waitingTime) {
+            if (overrides.getWaitingTime() != null
+                && !overrides.getWaitingTime().isZero()
+                && !overrides.getWaitingTime().equals(this.waitingTime)
+                && !overrides.getWaitingTime().equals(DEFAULT_WAITING_TIME)
+            ) {
                 this.waitingTime = overrides.getWaitingTime();
             }
         }

diff --git a/libs/micronaut-worker/src/main/java/com/agorapulse/worker/convention/QueueConsumer.java b/libs/micronaut-worker/src/main/java/com/agorapulse/worker/convention/QueueConsumer.java
@@ -36,7 +36,7 @@
 @Documented
 @Consumes
 @Fork(JobConfiguration.ConsumerQueueConfiguration.DEFAULT_MAX_MESSAGES)
-@FixedRate("20s")
+@FixedRate(JobConfiguration.ConsumerQueueConfiguration.DEFAULT_WAITING_TIME_STRING)
 @Retention(RUNTIME)
 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
 public @interface QueueConsumer {
@@ -70,7 +70,7 @@
      */
     @AliasFor(annotation = Consumes.class, member = "waitingTime")
     @AliasFor(annotation = FixedRate.class, member = "value")
-    String waitingTime() default "";
+    String waitingTime() default JobConfiguration.ConsumerQueueConfiguration.DEFAULT_WAITING_TIME_STRING;
 
     /**
      * The number of messages to consume and also the number of threads to use to consume the messages.