aboutsummaryrefslogtreecommitdiff
path: root/src/main/java/io/trygvis/queue
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/io/trygvis/queue')
-rw-r--r--src/main/java/io/trygvis/queue/QueueExecutor.java8
-rw-r--r--src/main/java/io/trygvis/queue/QueueService.java26
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;
+ }
}
}