You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2021/10/29 11:11:20 UTC
[camel] 01/04: CAMEL-17121: create a wrapper for repeatable tasks
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 987691f67f84d6db33fcf993be1e31a7634e54ea
Author: Otavio R. Piske <an...@gmail.com>
AuthorDate: Sun Oct 24 15:00:13 2021 +0200
CAMEL-17121: create a wrapper for repeatable tasks
This provides the reusable API to replace multiple do-test-wait-and-retry loops
in component code
---
core/camel-support/pom.xml | 20 +++
.../camel/support/task/AbstractTaskBuilder.java | 44 ++++++
.../apache/camel/support/task/BackgroundTask.java | 156 ++++++++++++++++++++
.../apache/camel/support/task/BlockingTask.java | 50 +++++++
.../apache/camel/support/task/ForegroundTask.java | 123 ++++++++++++++++
.../java/org/apache/camel/support/task/Task.java | 25 ++++
.../org/apache/camel/support/task/TaskBuilder.java | 33 +++++
.../java/org/apache/camel/support/task/Tasks.java | 60 ++++++++
.../apache/camel/support/task/budget/Budget.java | 37 +++++
.../camel/support/task/budget/BudgetBuilder.java | 32 +++++
.../apache/camel/support/task/budget/Budgets.java | 91 ++++++++++++
.../task/budget/IterationBoundedBudget.java | 66 +++++++++
.../task/budget/IterationBoundedBudgetBuilder.java | 57 ++++++++
.../camel/support/task/budget/IterationBudget.java | 52 +++++++
.../task/budget/IterationTimeBoundedBudget.java | 84 +++++++++++
.../budget/IterationTimeBoundedBudgetBuilder.java | 65 +++++++++
.../support/task/budget/TimeBoundedBudget.java | 45 ++++++
.../camel/support/task/budget/TimeBudget.java | 31 ++++
.../camel/support/task/BackgroundTaskTest.java | 146 +++++++++++++++++++
.../camel/support/task/ForegroundTaskTest.java | 132 +++++++++++++++++
.../camel/support/task/ForegroundTimeTaskTest.java | 157 +++++++++++++++++++++
.../apache/camel/support/task/TaskTestSupport.java | 74 ++++++++++
.../src/test/resources/log4j2.properties | 32 +++++
23 files changed, 1612 insertions(+)
diff --git a/core/camel-support/pom.xml b/core/camel-support/pom.xml
index b03750e..db3c277 100644
--- a/core/camel-support/pom.xml
+++ b/core/camel-support/pom.xml
@@ -51,6 +51,26 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+
+ <!-- testing -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <version>${assertj-version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTaskBuilder.java b/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTaskBuilder.java
new file mode 100644
index 0000000..eaa197d
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/AbstractTaskBuilder.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+/**
+ * Provides common logic for task builders
+ *
+ * @param <T> the type of the task
+ */
+public abstract class AbstractTaskBuilder<T extends Task> implements TaskBuilder<T> {
+ protected static final String DEFAULT_NAME = "camel-repeatable-task";
+ private String name = AbstractTaskBuilder.DEFAULT_NAME;
+
+ /**
+ * Assigns a name to the task being built
+ *
+ * @param name the name of the task
+ * @return A reference to this object
+ */
+ public AbstractTaskBuilder<T> withName(String name) {
+ this.name = name;
+
+ return this;
+ }
+
+ protected String getName() {
+ return name;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
new file mode 100644
index 0000000..a718aff
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.function.Predicate;
+
+import org.apache.camel.support.task.budget.TimeBudget;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A sleepless blocking task that runs in a Thread the background. The execution is blocked until the task budget is
+ * exhausted. All background tasks are constrained by a time budget.
+ */
+public class BackgroundTask implements BlockingTask {
+
+ /**
+ * A builder helper for building new background tasks
+ */
+ public static class BackgroundTaskBuilder extends AbstractTaskBuilder<BackgroundTask> {
+ private TimeBudget budget;
+ private ScheduledExecutorService service;
+
+ /**
+ * Sets a time budget for the task
+ *
+ * @param timeBudget the time budget
+ * @return
+ */
+ public BackgroundTaskBuilder withBudget(TimeBudget timeBudget) {
+ this.budget = timeBudget;
+
+ return this;
+ }
+
+ /**
+ * Sets a executor service manager for managing the threads
+ *
+ * @param service an instance of an executor service to use
+ * @return
+ */
+ public BackgroundTaskBuilder withScheduledExecutor(ScheduledExecutorService service) {
+ this.service = service;
+
+ return this;
+ }
+
+ @Override
+ public BackgroundTask build() {
+ return new BackgroundTask(budget, service, getName());
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BackgroundTask.class);
+
+ private final TimeBudget budget;
+ private final ScheduledExecutorService service;
+ private final String name;
+
+ BackgroundTask(TimeBudget budget, ScheduledExecutorService service, String name) {
+ this.budget = budget;
+ this.service = Objects.requireNonNull(service);
+ this.name = name;
+ }
+
+ private <T> void runTaskWrapper(CountDownLatch latch, Predicate<T> predicate, T payload) {
+ LOG.trace("Current latch value: {}", latch.getCount());
+
+ if (latch.getCount() == 0) {
+ return;
+ }
+
+ if (predicate.test(payload)) {
+ latch.countDown();
+ LOG.trace("Task {} has succeeded and the current task won't be schedulable anymore: {}", name, latch.getCount());
+ }
+ }
+
+ private void runTaskWrapper(CountDownLatch latch, BooleanSupplier supplier) {
+ LOG.trace("Current latch value: {}", latch.getCount());
+ if (latch.getCount() == 0) {
+ return;
+ }
+
+ if (supplier.getAsBoolean()) {
+ latch.countDown();
+ LOG.trace("Task {} succeeded and the current task won't be schedulable anymore: {}", name, latch.getCount());
+ }
+ }
+
+ @Override
+ public <T> boolean run(Predicate<T> predicate, T payload) {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ // We need it to be cancellable/non-runnable after reaching a certain point, and it needs to be deterministic.
+ // This is why we ignore the ScheduledFuture returned and implement the go/no-go using a latch.
+ service.scheduleAtFixedRate(() -> runTaskWrapper(latch, predicate, payload),
+ budget.initialDelay(), budget.interval(), TimeUnit.MILLISECONDS);
+
+ return waitForTaskCompletion(latch, service);
+ }
+
+ @Override
+ public boolean run(BooleanSupplier supplier) {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ // We need it to be cancellable/non-runnable after reaching a certain point, and it needs to be deterministic.
+ // This is why we ignore the ScheduledFuture returned and implement the go/no-go using a latch.
+ service.scheduleAtFixedRate(() -> runTaskWrapper(latch, supplier), budget.initialDelay(),
+ budget.interval(), TimeUnit.MILLISECONDS);
+
+ return waitForTaskCompletion(latch, service);
+ }
+
+ private boolean waitForTaskCompletion(CountDownLatch latch, ScheduledExecutorService service) {
+ boolean completed = false;
+ try {
+ if (!latch.await(budget.maxDuration(), TimeUnit.MILLISECONDS)) {
+ LOG.debug("Timeout out waiting for the completion of the task");
+ } else {
+ LOG.info("The task is complete after iterations and the code is ready to continue");
+
+ completed = true;
+ }
+
+ service.shutdown();
+ service.awaitTermination(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while waiting for the repeatable task to execute");
+ } finally {
+ service.shutdownNow();
+ }
+
+ return completed;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java b/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
new file mode 100644
index 0000000..a7e93e1
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.util.function.BooleanSupplier;
+import java.util.function.Predicate;
+
+/**
+ * Defines a task that blocks the code execution when ran. The task under execution must be thread-safe.
+ */
+public interface BlockingTask extends Task {
+ /**
+ * Run the task
+ *
+ * @param predicate the task as a predicate. The result of the predicate is used to check if the task has completed
+ * or not. The predicate must return true if the execution has completed or false otherwise.
+ * Failures on the task should be handled on the predicate using the payload as wrapper for In/Out
+ * if necessary
+ * @param payload a payload to be passed to the task
+ * @param <T> The type of the payload passed to the predicate when testing the task
+ * @return true if the task has completed successfully or false if: 1) the budget is exhausted or 2) the
+ * task was interrupted.
+ */
+ <T> boolean run(Predicate<T> predicate, T payload);
+
+ /**
+ * Run the task
+ *
+ * @param supplier the task as a boolean supplier. The result is used to check if the task has completed or not.
+ * The supplier must return true if the execution has completed or false otherwise.
+ * @return true if the task has completed successfully or false if: 1) the budget is exhausted or 2) the
+ * task was interrupted.
+ */
+ boolean run(BooleanSupplier supplier);
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
new file mode 100644
index 0000000..f7edb36
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.util.function.BooleanSupplier;
+import java.util.function.Predicate;
+
+import org.apache.camel.support.task.budget.IterationBudget;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Runs a task in the foreground, executing for a given number of iteration and sleeping between each of them.
+ */
+public class ForegroundTask implements BlockingTask {
+
+ /**
+ * A builder helper for building new foreground tasks
+ */
+ public static class ForegroundTaskBuilder extends AbstractTaskBuilder<ForegroundTask> {
+ private IterationBudget budget;
+
+ /**
+ * Sets an iteration budget for the task (i.e.: the task will not run more than the given number of iterations)
+ *
+ * @param budget the budget
+ * @return an instance of the this builder
+ */
+ public ForegroundTaskBuilder withBudget(IterationBudget budget) {
+ this.budget = budget;
+
+ return this;
+ }
+
+ @Override
+ public ForegroundTask build() {
+ return new ForegroundTask(budget, getName());
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(ForegroundTask.class);
+
+ private final String name;
+ private IterationBudget budget;
+
+ ForegroundTask(IterationBudget budget, String name) {
+ this.budget = budget;
+ this.name = name;
+ }
+
+ @Override
+ public <T> boolean run(Predicate<T> predicate, T payload) {
+ boolean completed = false;
+ try {
+ if (budget.initialDelay() > 0) {
+ Thread.sleep(budget.initialDelay());
+ }
+
+ while (budget.next()) {
+ if (predicate.test(payload)) {
+ LOG.info("Task {} is complete after {} iterations and it is ready to continue",
+ name, budget.iterations());
+ completed = true;
+ break;
+ }
+
+ if (budget.canContinue()) {
+ Thread.sleep(budget.interval());
+ }
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted {} while waiting for the repeatable task to execute", name);
+ Thread.currentThread().interrupt();
+ }
+
+ return completed;
+ }
+
+ @Override
+ public boolean run(BooleanSupplier supplier) {
+ boolean completed = false;
+
+ try {
+ if (budget.initialDelay() > 0) {
+ Thread.sleep(budget.initialDelay());
+ }
+
+ while (budget.next()) {
+ if (supplier.getAsBoolean()) {
+ LOG.info("Task {} is complete after {} iterations and it is ready to continue",
+ name, budget.iterations());
+ completed = true;
+
+ break;
+ }
+
+ if (budget.canContinue()) {
+ Thread.sleep(budget.interval());
+ }
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted {} while waiting for the repeatable task to execute", name);
+ Thread.currentThread().interrupt();
+ }
+
+ return completed;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java b/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java
new file mode 100644
index 0000000..e6771b3
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/Task.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+/**
+ * A task defines a piece of code that may be executed - in the foreground or background - within a certain budget that
+ * is specific to the task type.
+ */
+public interface Task {
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/TaskBuilder.java b/core/camel-support/src/main/java/org/apache/camel/support/task/TaskBuilder.java
new file mode 100644
index 0000000..7fb09ea
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/TaskBuilder.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+/**
+ * Provides a common pattern for building tasks
+ *
+ * @param <T> the task type to build
+ */
+public interface TaskBuilder<T extends Task> {
+
+ /**
+ * Builds the task of the given type for the builder
+ *
+ * @return a new instance of the task
+ */
+ T build();
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/Tasks.java b/core/camel-support/src/main/java/org/apache/camel/support/task/Tasks.java
new file mode 100644
index 0000000..b95c3a7
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/Tasks.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.util.function.Supplier;
+
+/**
+ * A helper class for building tasks
+ */
+public final class Tasks {
+
+ private Tasks() {
+
+ }
+
+ /**
+ * Creates a new background task builder
+ *
+ * @return an instance of a background task builder
+ */
+ public static BackgroundTask.BackgroundTaskBuilder backgroundTask() {
+ return task(BackgroundTask.BackgroundTaskBuilder::new);
+ }
+
+ /**
+ * Creates a new foreground task builder
+ *
+ * @return an instance of a foreground task builder
+ */
+ public static ForegroundTask.ForegroundTaskBuilder foregroundTask() {
+ return task(ForegroundTask.ForegroundTaskBuilder::new);
+ }
+
+ /**
+ * A generic builder for task builders
+ *
+ * @param taskBuilderSupplier A supplier of tasks (usually a parameterless constructor in the form of Builder::new)
+ * @param <T> the type of tasks that the builder builds
+ * @param <Y> the type of the task builder to provide
+ * @return A new instance of the given task builder
+ */
+ public static <T extends BlockingTask, Y extends TaskBuilder<T>> Y task(Supplier<Y> taskBuilderSupplier) {
+ return taskBuilderSupplier.get();
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budget.java
new file mode 100644
index 0000000..fc50da5
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budget.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+/**
+ * A budget defines how much a task can execute
+ */
+public interface Budget {
+ /**
+ * Defines an initial delay before running the task
+ *
+ * @return the initial delay, in milliseconds, before running the task
+ */
+ long initialDelay();
+
+ /**
+ * The interval between each task execution
+ *
+ * @return the interval, in milliseconds, for each task execution
+ */
+ long interval();
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/BudgetBuilder.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/BudgetBuilder.java
new file mode 100644
index 0000000..bf77d0b
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/BudgetBuilder.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+/**
+ * An interface for budget builders
+ *
+ * @param <T>
+ */
+public interface BudgetBuilder<T extends Budget> {
+ /**
+ * Build the budget
+ *
+ * @return
+ */
+ T build();
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budgets.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budgets.java
new file mode 100644
index 0000000..21c05dc
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/Budgets.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+import java.time.Duration;
+
+/**
+ * Helper builder of budgets
+ */
+public final class Budgets {
+ private static final long DEFAULT_INITIAL_DELAY = 0;
+ private static final long DEFAULT_INTERVAL = 1000;
+
+ /**
+ * A helper builder of time bounded builders. Provide generic/safe default values, but should be adjusted on a
+ * per-case basis
+ */
+ public static class TimeBoundedBudgetBuilder {
+ private static final long DEFAULT_MAX_DURATION = 5000;
+
+ private long initialDelay = DEFAULT_INITIAL_DELAY;
+ private long interval = DEFAULT_INTERVAL;
+ private long maxDuration = DEFAULT_MAX_DURATION;
+
+ public TimeBoundedBudgetBuilder withInitialDelay(Duration duration) {
+ this.initialDelay = duration.toMillis();
+
+ return this;
+ }
+
+ public TimeBoundedBudgetBuilder withInterval(Duration duration) {
+ this.interval = duration.toMillis();
+
+ return this;
+ }
+
+ public TimeBoundedBudgetBuilder withMaxDuration(Duration duration) {
+ this.maxDuration = duration.toMillis();
+
+ return this;
+ }
+
+ public TimeBoundedBudget build() {
+ return new TimeBoundedBudget(initialDelay, interval, maxDuration);
+ }
+ }
+
+ private Budgets() {
+ }
+
+ public static TimeBoundedBudgetBuilder timeBudget() {
+ return new TimeBoundedBudgetBuilder();
+ }
+
+ public static IterationBoundedBudgetBuilder iterationBudget() {
+ return new IterationBoundedBudgetBuilder();
+ }
+
+ public static IterationTimeBoundedBudgetBuilder iterationTimeBudget() {
+ return new IterationTimeBoundedBudgetBuilder();
+ }
+
+ /**
+ * Some components use 0 to disable retring the task. This sanitizes it to run at least once
+ *
+ * @param iterations the number of iterations
+ * @return an integer greater than or equal to 1 equivalent to the maximum number of iterations allowed
+ */
+ public static int atLeastOnce(int iterations) {
+ if (iterations <= 0) {
+ return 1;
+ }
+
+ return iterations;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
new file mode 100644
index 0000000..86ccee6
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudget.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+public class IterationBoundedBudget implements IterationBudget {
+ private final long initialDelay;
+ private final long interval;
+ private final int maxIterations;
+ private int iterations;
+
+ IterationBoundedBudget(long initialDelay, long interval, int maxIterations) {
+ this.initialDelay = initialDelay;
+ this.interval = interval;
+ this.maxIterations = maxIterations;
+ }
+
+ @Override
+ public long initialDelay() {
+ return initialDelay;
+ }
+
+ @Override
+ public long interval() {
+ return interval;
+ }
+
+ @Override
+ public int maxIterations() {
+ return maxIterations;
+ }
+
+ public int iterations() {
+ return iterations;
+ }
+
+ @Override
+ public boolean next() {
+ if (canContinue()) {
+ iterations++;
+
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean canContinue() {
+ return iterations < maxIterations;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudgetBuilder.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudgetBuilder.java
new file mode 100644
index 0000000..4770b29
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBoundedBudgetBuilder.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+import java.time.Duration;
+
+/**
+ * A helper builder of iteration bounded builders. Provide generic/safe default values, but should be adjusted on a
+ * per-case basis. By default, execute the iterations for up to Integer.MAX_VALUE.
+ */
+public class IterationBoundedBudgetBuilder implements BudgetBuilder<IterationBudget> {
+ private static final int DEFAULT_MAX_ITERATIONS = Integer.MAX_VALUE;
+ private static final long DEFAULT_INITIAL_DELAY = 0;
+ private static final long DEFAULT_INTERVAL = 1000;
+
+ protected long initialDelay = DEFAULT_INITIAL_DELAY;
+ protected long interval = DEFAULT_INTERVAL;
+ protected int maxIterations = DEFAULT_MAX_ITERATIONS;
+
+ public IterationBoundedBudgetBuilder withInitialDelay(Duration duration) {
+ this.initialDelay = duration.toMillis();
+
+ return this;
+ }
+
+ public IterationBoundedBudgetBuilder withInterval(Duration duration) {
+ this.interval = duration.toMillis();
+
+ return this;
+ }
+
+ public IterationBoundedBudgetBuilder withMaxIterations(int maxIterations) {
+ this.maxIterations = maxIterations;
+
+ return this;
+ }
+
+ @Override
+ public IterationBoundedBudget build() {
+ return new IterationBoundedBudget(initialDelay, interval, maxIterations);
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBudget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBudget.java
new file mode 100644
index 0000000..d01e74e
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationBudget.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+/**
+ * A budget that limits the task execution to a given number of iterations
+ */
+public interface IterationBudget extends Budget {
+
+ /**
+ * The maximum number of iterations
+ *
+ * @return
+ */
+ int maxIterations();
+
+ /**
+ * The current number of iterations
+ *
+ * @return the current number of iterations
+ */
+ int iterations();
+
+ /**
+ * Whether the task has budget to continue executing or not
+ *
+ * @return true if the task can continue or false otherwise
+ */
+ boolean canContinue();
+
+ /**
+ * Move the task to the next iteration
+ *
+ * @return true if the task can continue or false otherwise
+ */
+ boolean next();
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudget.java
new file mode 100644
index 0000000..ac24b32
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudget.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+import java.time.Duration;
+import java.time.Instant;
+
+public class IterationTimeBoundedBudget implements IterationBudget, TimeBudget {
+ private IterationBudget iterationBudget;
+ private final long maxDuration;
+ private final Instant startTime;
+
+ public IterationTimeBoundedBudget(long initialDelay, long interval, int maxIterations, long maxDuration) {
+ iterationBudget = new IterationBoundedBudget(initialDelay, interval, maxIterations);
+ this.maxDuration = maxDuration;
+ this.startTime = Instant.now();
+ }
+
+ @Override
+ public long initialDelay() {
+ return iterationBudget.initialDelay();
+ }
+
+ @Override
+ public long interval() {
+ return iterationBudget.interval();
+ }
+
+ @Override
+ public int maxIterations() {
+ return iterationBudget.maxIterations();
+ }
+
+ @Override
+ public int iterations() {
+ return iterationBudget.maxIterations();
+ }
+
+ @Override
+ public boolean next() {
+ if (canContinue()) {
+ return iterationBudget.next();
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean canContinue() {
+ // Abort if the iteration budget or ...
+ if (!iterationBudget.canContinue()) {
+ return false;
+ }
+
+ // ... the time budget is exhausted
+ if (Duration.between(startTime, Instant.now()).toMillis() >= maxDuration) {
+ return false;
+ }
+
+ // Otherwise, can continue to schedule/run the task
+ return true;
+ }
+
+ @Override
+ public long maxDuration() {
+ return maxDuration;
+ }
+
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudgetBuilder.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudgetBuilder.java
new file mode 100644
index 0000000..7c07d50
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/IterationTimeBoundedBudgetBuilder.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+import java.time.Duration;
+
+/**
+ * A helper builder of iteration and time bounded builders. Provide generic/safe default values, but should be adjusted
+ * on a per-case basis. By default, execute the iterations for up to Integer.MAX_VALUE and a duration of 5 seconds.
+ */
+public class IterationTimeBoundedBudgetBuilder implements BudgetBuilder<IterationTimeBoundedBudget> {
+ private static final long DEFAULT_MAX_DURATION = 5000;
+ private static final int DEFAULT_MAX_ITERATIONS = Integer.MAX_VALUE;
+ private static final long DEFAULT_INITIAL_DELAY = 0;
+ private static final long DEFAULT_INTERVAL = 1000;
+
+ private long maxDuration = DEFAULT_MAX_DURATION;
+ private long initialDelay = DEFAULT_INITIAL_DELAY;
+ private long interval = DEFAULT_INTERVAL;
+ private int maxIterations = DEFAULT_MAX_ITERATIONS;
+
+ public IterationTimeBoundedBudgetBuilder withInitialDelay(Duration duration) {
+ this.initialDelay = duration.toMillis();
+
+ return this;
+ }
+
+ public IterationTimeBoundedBudgetBuilder withInterval(Duration duration) {
+ this.interval = duration.toMillis();
+
+ return this;
+ }
+
+ public IterationTimeBoundedBudgetBuilder withMaxIterations(int maxIterations) {
+ this.maxIterations = maxIterations;
+
+ return this;
+ }
+
+ public IterationTimeBoundedBudgetBuilder withMaxDuration(Duration duration) {
+ this.maxDuration = duration.toMillis();
+
+ return this;
+ }
+
+ @Override
+ public IterationTimeBoundedBudget build() {
+ return new IterationTimeBoundedBudget(initialDelay, interval, maxIterations, maxDuration);
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
new file mode 100644
index 0000000..bb01b0b
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBoundedBudget.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+public class TimeBoundedBudget implements TimeBudget {
+ private final long initialDelay;
+ private final long interval;
+ private final long maxDuration;
+
+ TimeBoundedBudget(long initialDelay, long interval, long maxDuration) {
+ this.initialDelay = initialDelay;
+ this.interval = interval;
+ this.maxDuration = maxDuration;
+ }
+
+ @Override
+ public long maxDuration() {
+ return maxDuration;
+ }
+
+ @Override
+ public long initialDelay() {
+ return initialDelay;
+ }
+
+ @Override
+ public long interval() {
+ return interval;
+ }
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBudget.java b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBudget.java
new file mode 100644
index 0000000..430fff3
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/task/budget/TimeBudget.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task.budget;
+
+/**
+ * Defines a budge that limits the task execution to certain length of time
+ */
+public interface TimeBudget extends Budget {
+
+ /**
+ * The maximum length of time that the task can execute
+ *
+ * @return the length of time in milliseconds
+ */
+ long maxDuration();
+}
diff --git a/core/camel-support/src/test/java/org/apache/camel/support/task/BackgroundTaskTest.java b/core/camel-support/src/test/java/org/apache/camel/support/task/BackgroundTaskTest.java
new file mode 100644
index 0000000..c5a4bb9
--- /dev/null
+++ b/core/camel-support/src/test/java/org/apache/camel/support/task/BackgroundTaskTest.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+
+import org.apache.camel.support.task.budget.Budgets;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BackgroundTaskTest extends TaskTestSupport {
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier")
+ @Test
+ @Timeout(10)
+ void testRunNoMoreSupplier() {
+ // It should run 5 times in 4 seconds because there is no delay
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ZERO)
+ .withMaxDuration(Duration.ofSeconds(4))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ assertFalse(completed, "The task did not complete, the return should be false");
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier")
+ @Test
+ @Timeout(10)
+ void testRunNoMoreSupplierWithDelay() {
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withMaxDuration(Duration.ofSeconds(5))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ assertFalse(completed, "The task did not complete, the return should be false");
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a predicate and an initial delay")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicate() {
+ // It should run 5 times in 4 seconds because there is no delay
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ZERO)
+ .withMaxDuration(Duration.ofSeconds(4))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::taskPredicate, new Object());
+ assertEquals(maxIterations, taskCount);
+ assertFalse(completed, "The task did not complete, the return should be false");
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministic() {
+ // It should run 5 times in 4 seconds because there is no delay
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ZERO)
+ .withMaxDuration(Duration.ofSeconds(4))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::taskPredicateWithDeterministicStop, Integer.valueOf(3));
+ assertEquals(3, taskCount);
+ assertTrue(completed, "The task did complete, the return should be true");
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlow() {
+ // It should run 5 times in 4 seconds because there is no delay
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ZERO)
+ .withMaxDuration(Duration.ofSeconds(4))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ assertFalse(completed, "The task did not complete because of timeout, the return should be false");
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlowWithDelay() {
+ // It should run 5 times in 4 seconds because there is no delay
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(Executors.newSingleThreadScheduledExecutor())
+ .withBudget(Budgets.timeBudget()
+ .withInterval(Duration.ofSeconds(1))
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withMaxDuration(Duration.ofSeconds(4))
+ .build())
+ .build();
+
+ boolean completed = task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ assertFalse(completed, "The task did not complete because of timeout, the return should be false");
+ }
+}
diff --git a/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTaskTest.java b/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTaskTest.java
new file mode 100644
index 0000000..7b69371
--- /dev/null
+++ b/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTaskTest.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.time.Duration;
+
+import org.apache.camel.support.task.budget.Budgets;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ForegroundTaskTest extends TaskTestSupport {
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier")
+ @Test
+ // @Timeout(10)
+ void testRunNoMoreSupplier() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier")
+ @Test
+ @Timeout(10)
+ void testRunNoMoreSupplierWithDelay() {
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a predicate and an initial delay")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicate() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicate, new Object());
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministic() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStop, Integer.valueOf(3));
+ assertEquals(3, taskCount);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlow() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlowWithDelay() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ }
+}
diff --git a/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTimeTaskTest.java b/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTimeTaskTest.java
new file mode 100644
index 0000000..a8b0139
--- /dev/null
+++ b/core/camel-support/src/test/java/org/apache/camel/support/task/ForegroundTimeTaskTest.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import java.time.Duration;
+
+import org.apache.camel.support.task.budget.Budgets;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ForegroundTimeTaskTest extends TaskTestSupport {
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier")
+ @Test
+ @Timeout(10)
+ void testRunNoMoreSupplier() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a supplier and an initial delay")
+ @Test
+ @Timeout(10)
+ void testRunNoMoreSupplierWithDelay() {
+ // this should run 5 times in a total duration of 6 seconds (5s executing + 1s delay)
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(6))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::booleanSupplier);
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task does not run for more than the max iterations when using a predicate and an initial delay")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicate() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicate, new Object());
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task does not run for more than the max duration when using a predicate and an initial delay")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateMaxDuration() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(50)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicate, new Object());
+ assertEquals(maxIterations, taskCount);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministic() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStop, Integer.valueOf(3));
+ assertEquals(3, taskCount);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlow() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ZERO)
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ }
+
+ @DisplayName("Test that the task stops running once the predicate is true when the test is slow")
+ @Test
+ @Timeout(10)
+ void testRunNoMorePredicateDeterministicSlowWithDelay() {
+ // It should run 5 times in 4 seconds because there is no delay
+ ForegroundTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofSeconds(5))
+ .withMaxIterations(5)
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withInterval(Duration.ofSeconds(1))
+ .build())
+ .build();
+
+ task.run(this::taskPredicateWithDeterministicStopSlow, Integer.valueOf(3));
+ assertTrue(taskCount < maxIterations);
+ }
+}
diff --git a/core/camel-support/src/test/java/org/apache/camel/support/task/TaskTestSupport.java b/core/camel-support/src/test/java/org/apache/camel/support/task/TaskTestSupport.java
new file mode 100644
index 0000000..7396a95
--- /dev/null
+++ b/core/camel-support/src/test/java/org/apache/camel/support/task/TaskTestSupport.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.support.task;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class TaskTestSupport {
+ protected final int maxIterations = 5;
+ protected volatile int taskCount;
+
+ protected boolean booleanSupplier() {
+ taskCount++;
+
+ return false;
+ }
+
+ protected boolean taskPredicate(Object o) {
+ assertNotNull(o);
+ taskCount++;
+
+ return false;
+ }
+
+ protected boolean taskPredicateWithDeterministicStop(Integer stopAtValue) {
+ assertNotNull(stopAtValue);
+
+ taskCount++;
+ if (taskCount == stopAtValue.intValue()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ protected boolean taskPredicateWithDeterministicStopSlow(Integer stopAtValue) {
+ assertNotNull(stopAtValue);
+
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ System.out.println("Interrupted");
+ return false;
+ }
+
+ taskCount++;
+ if (taskCount == stopAtValue.intValue()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ @BeforeEach
+ void setUp() {
+ taskCount = 0;
+ }
+}
diff --git a/core/camel-support/src/test/resources/log4j2.properties b/core/camel-support/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..2f560dd
--- /dev/null
+++ b/core/camel-support/src/test/resources/log4j2.properties
@@ -0,0 +1,32 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+appender.console.type = Console
+appender.console.name = console
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-support-test.log
+appender.file.append = true
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+rootLogger.level = INFO
+
+rootLogger.appenderRef.file.ref = file
+#rootLogger.appenderRef.console.ref = console