Merge pull request #52 from agorapulse/chore/align-forsk-with-number-of-messages

align forks with number of messages

Author: musketyr

docs/guide/src/docs/asciidoc/usage.adoc

@@ -138,6 +138,8 @@ include::{root-dir}/libs/micronaut-worker/src/test/groovy/com/agorapulse/worker/
 <5> Use the `@QueueConsumer` annotation with the name of the queue to consume messages
 <6> The consumer job must have a single parameter of the same type as the producer job returns
 
+TIP: The value for `@Fork` is the same as the number of `maxMessages` for `@QueueConsumer` annotation. The value of `waitingTime` in `@QueueConsumer` is the same as the associated `@FixedRate` value.
+
 ==== Advanced Usage
 
 The name of the queue can be customised using `@Consumes` and `@Produces`

libs/micronaut-worker-executor-redis/micronaut-worker-executor-redis.gradle

@@ -24,4 +24,5 @@ dependencies {
     implementation 'io.micronaut.reactor:micronaut-reactor'
 
     testImplementation project(':micronaut-worker-tck')
+    testImplementation 'org.yaml:snakeyaml'
 }

libs/micronaut-worker-executor-redis/src/test/groovy/com/agorapulse/worker/redis/RedisJobExecutorSpec.groovy

@@ -45,17 +45,7 @@ class RedisJobExecutorSpec extends AbstractJobExecutorSpec {
 
         ApplicationContext ctx = ApplicationContext
                 .builder(
-                        'redis.uri': "redis://$redis.host:${redis.getMappedPort(6379)}",
-                        'worker.jobs.long-running-job-execute-producer.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-on-leader.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-on-follower.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-consecutive.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-unlimited.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-concurrent.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-concurrent-consumer.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-fork-consumer.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-regular-consumer.enabled': 'true',
-                        'worker.jobs.long-running-job-execute-fork.enabled': 'true'
+                        'redis.uri': "redis://$redis.host:${redis.getMappedPort(6379)}"
                 )
                 .environments(CONCURRENT_JOB_TEST_ENVIRONMENT)
                 .build()

libs/micronaut-worker-executor-redis/src/test/resources/application.yml

@@ -0,0 +1,57 @@
+
+
+worker:
+  jobs:
+    long-running-job-execute-producer:
+      enabled: true
+      scheduler: long-running-job-execute-producer
+    long-running-job-execute-on-leader:
+      enabled: true
+      scheduler: long-running-job-execute-on-leader
+    long-running-job-execute-on-follower:
+      enabled: true
+      scheduler: long-running-job-execute-on-follower
+    long-running-job-execute-consecutive:
+      enabled: true
+      scheduler: long-running-job-execute-consecutive
+    long-running-job-execute-unlimited:
+      enabled: true
+      scheduler: long-running-job-execute-unlimited
+    long-running-job-execute-concurrent:
+      enabled: true
+      scheduler: long-running-job-execute-concurrent
+    long-running-job-execute-concurrent-consumer:
+      enabled: true
+      scheduler: long-running-job-execute-concurrent-consumer
+    long-running-job-execute-fork-consumer:
+      enabled: true
+      scheduler: long-running-job-execute-fork-consumer
+    long-running-job-execute-regular-consumer:
+      enabled: true
+      scheduler: long-running-job-execute-regular-consumer
+    long-running-job-execute-fork:
+      enabled: true
+      scheduler: long-running-job-execute-fork
+
+micronaut:
+  executors:
+    long-running-job-execute-producer:
+      number-of-threads: 1
+    long-running-job-execute-on-leader:
+        number-of-threads: 1
+    long-running-job-execute-on-follower:
+        number-of-threads: 1
+    long-running-job-execute-consecutive:
+        number-of-threads: 1
+    long-running-job-execute-unlimited:
+        number-of-threads: 1
+    long-running-job-execute-concurrent:
+        number-of-threads: 1
+    long-running-job-execute-concurrent-consumer:
+        number-of-threads: 1
+    long-running-job-execute-fork-consumer:
+        number-of-threads: 2
+    long-running-job-execute-regular-consumer:
+        number-of-threads: 1
+    long-running-job-execute-fork:
+        number-of-threads: 2

libs/micronaut-worker/src/main/java/com/agorapulse/worker/convention/QueueConsumer.java

@@ -21,6 +21,7 @@
 import com.agorapulse.worker.WorkerConfiguration;
 import com.agorapulse.worker.annotation.Consumes;
 import com.agorapulse.worker.annotation.FixedRate;
+import com.agorapulse.worker.annotation.Fork;
 import com.agorapulse.worker.annotation.Job;
 import io.micronaut.context.annotation.AliasFor;
 import jakarta.inject.Named;
@@ -34,6 +35,7 @@
 
 @Documented
 @Consumes
+@Fork(JobConfiguration.ConsumerQueueConfiguration.DEFAULT_MAX_MESSAGES)
 @FixedRate("20s")
 @Retention(RUNTIME)
 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@@ -61,7 +63,8 @@
     /**
      * @return the maximum of messages consumed in a single run, defaults to {@link JobConfiguration.ConsumerQueueConfiguration#DEFAULT_MAX_MESSAGES}
      */
-    @AliasFor(annotation = Consumes.class, member = "value")
+    @AliasFor(annotation = Fork.class, member = "value")
+    @AliasFor(annotation = Consumes.class, member = "maxMessages")
     int maxMessages() default JobConfiguration.ConsumerQueueConfiguration.DEFAULT_MAX_MESSAGES;
 
     /**