o Adding an ActiveMQ connection to messages can be sent as a hint to the consumer.HEADmaster

2 files changed, 29 insertions, 5 deletions

diff --git a/src/main/java/io/trygvis/queue/QueueExecutor.java b/src/main/java/io/trygvis/queue/QueueExecutor.java
index 468059d..88e5b46 100644
--- a/src/main/java/io/trygvis/queue/QueueExecutor.java
+++ b/src/main/java/io/trygvis/queue/QueueExecutor.java
@@ -40,9 +40,10 @@ public class QueueExecutor {
         public int ok;
         public int failed;
         public int scheduled;
+        public int missed;
 
-        public QueueStats toStats() {
-            return new QueueStats(total, ok, failed, scheduled);
+        public synchronized QueueStats toStats() {
+            return new QueueStats(total, ok, failed, missed, scheduled);
         }
     }
 
@@ -94,6 +95,9 @@ public class QueueExecutor {
 
             if (count == 0) {
                 log.warn("Missed task {}", task.id());
+                synchronized (stats) {
+                    stats.missed++;
+                }
                 return MISSED;
             }
 
diff --git a/src/main/java/io/trygvis/queue/QueueService.java b/src/main/java/io/trygvis/queue/QueueService.java
index f4ce536..1c38f1f 100644
--- a/src/main/java/io/trygvis/queue/QueueService.java
+++ b/src/main/java/io/trygvis/queue/QueueService.java
@@ -12,15 +12,27 @@ public interface QueueService {
     public static class TaskExecutionRequest {
         public final long chunkSize;
         public final boolean stopOnError;
+        public final Long interval;
+        public final boolean continueOnFullChunk;
         // TODO: saveExceptions
 
         public TaskExecutionRequest(long chunkSize, boolean stopOnError) {
-            if (chunkSize <= 0) {
-                throw new IllegalArgumentException("chunkSize has to be bigger than zero.");
-            }
+            this(chunkSize, stopOnError, null, true);
+        }
 
+        private TaskExecutionRequest(long chunkSize, boolean stopOnError, Long interval, boolean continueOnFullChunk) {
             this.chunkSize = chunkSize;
             this.stopOnError = stopOnError;
+            this.interval = interval;
+            this.continueOnFullChunk = continueOnFullChunk;
+        }
+
+        public TaskExecutionRequest interval(long interval) {
+            return new TaskExecutionRequest(chunkSize, stopOnError, interval, continueOnFullChunk);
+        }
+
+        public TaskExecutionRequest continueOnFullChunk(boolean continueOnFullChunk) {
+            return new TaskExecutionRequest(chunkSize, stopOnError, interval, continueOnFullChunk);
         }
 
         @Override
@@ -30,5 +42,13 @@ public interface QueueService {
                     ", stopOnError=" + stopOnError +
                     '}';
         }
+
+        public long interval(Queue queue) {
+            if (interval != null) {
+                return interval;
+            }
+
+            return queue.interval;
+        }
     }
 }