You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2023/12/23 18:07:45 UTC
(camel-spring-boot-examples) branch main updated: Camel 20247 rework dynamic router component (#120)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-spring-boot-examples.git
The following commit(s) were added to refs/heads/main by this push:
new b328126 Camel 20247 rework dynamic router component (#120)
b328126 is described below
commit b328126fbb5c5dace68df98c310925a3168f5951
Author: Steve Storck <st...@gmail.com>
AuthorDate: Sat Dec 23 13:07:40 2023 -0500
Camel 20247 rework dynamic router component (#120)
* CAMEL-20247: Updated dynamic router spring boot example for component redesign.
* CAMEL-20247: Updated spring boot example for the Dynamic Router EIP Component to reflect component rework and for better demonstration use cases.
---
README.adoc | 2 +-
dynamic-router-eip/README.adoc | 285 +++++++++++----------
.../dynamic-router-eip-single/pom.xml | 25 ++
.../example/springboot/numbers}/Application.java | 2 +-
.../config/DynamicRouterComponentConfig.java | 157 ++++++++----
.../springboot/numbers/config/ExampleConfig.java | 50 ----
.../numbers/participants/PredicateConstants.java | 67 +++--
.../numbers/participants/RoutingParticipant.java | 25 +-
.../springboot/numbers/service/NumbersService.java | 43 ++--
.../springboot/numbers/service/ResultsService.java | 44 ++--
.../springboot/numbers/web/NumbersController.java | 22 ++
.../src/main/resources/application.yml | 15 +-
.../src/main/resources/logback.xml | 2 +-
.../all-numbers-service/pom.xml | 41 +++
.../springboot/numbers/all/Application.java | 33 +++
.../ProcessAllNumbersRoutingParticipant.java | 43 ++++
.../src/main/resources/application.yaml} | 67 ++---
.../even-numbers-service/pom.xml | 41 +++
.../springboot/numbers/even/Application.java | 31 +++
.../ProcessEvenNumbersRoutingParticipant.java | 42 +++
.../src/main/resources/application.yaml} | 67 ++---
.../dynamic-router-eip-stack/main-router/pom.xml | 68 +++++
.../springboot/numbers/mainrouter/Application.java | 31 +++
.../config/DynamicRouterComponentConfig.java | 39 +++
.../mainrouter/config/MainRouterConfig.java | 184 +++++++++++++
.../config/StateMachineActionConfig.java | 73 ++++++
.../mainrouter/config/StateMachineConfig.java | 89 +++++++
.../mainrouter/model/StateMachineEvent.java | 19 ++
.../NumberStatisticsRoutingParticipant.java | 75 ++++++
.../mainrouter/service/NumberGeneratorService.java | 70 +++++
.../service/NumberStatisticsService.java | 66 +++++
.../service/StateMachineEventTriggerService.java | 27 ++
.../mainrouter/service/SubscribersService.java | 25 ++
.../numbers/mainrouter/service/WarmUpService.java | 63 +++++
.../numbers/mainrouter/util/MainRouterUtil.java | 50 ++++
.../mainrouter/web/NumberGeneratorController.java | 33 +++
.../mainrouter/web/NumberStatisticsController.java | 40 +++
.../mainrouter/web/SubscribersController.java | 22 ++
.../src/main/resources/application.yaml} | 90 +++----
.../numbers-common/pom.xml | 45 ++++
.../numbers/common/config/CommonConfig.java | 37 +++
.../common/config/YamlPropertySourceFactory.java | 28 ++
.../numbers/common/model/CommandMessage.java | 34 +++
.../service/ProcessNumbersRoutingParticipant.java | 94 +++++++
.../numbers/common/service/RoutingParticipant.java | 166 ++++++++++++
.../service/StringValueHeaderDeserializer.java | 30 +++
.../numbers/common/util/NumbersCommonUtil.java | 25 ++
.../numbers-common/src/main/resources/common.yaml | 54 ++++
.../odd-numbers-service/pom.xml | 41 +++
.../springboot/numbers/odd/Application.java | 31 +++
.../ProcessOddNumbersRoutingParticipant.java | 42 +++
.../src/main/resources/application.yaml} | 67 ++---
.../dynamic-router-eip-stack/pom.xml | 44 ++++
.../project-resources/docker/docker-compose.yaml | 141 ++++++++++
dynamic-router-eip/pom.xml | 152 +++++++----
.../springboot/AllRecipientsApplicationTest.java | 16 --
.../springboot/FirstRecipientApplicationTest.java | 16 --
.../springboot/LessExpectedApplicationTest.java | 16 --
.../src/test/resources/logback-test.xml | 14 -
59 files changed, 2643 insertions(+), 618 deletions(-)
diff --git a/README.adoc b/README.adoc
index 3df272f..bcc8c13 100644
--- a/README.adoc
+++ b/README.adoc
@@ -79,7 +79,7 @@ Number of Examples: 54 (0 deprecated)
| link:arangodb/README.adoc[Arangodb] (arangodb) | Database | An example showing the Camel ArangoDb component with Spring Boot
-| link:dynamic-router-eip/README.adoc[Dynamic Router Eip] (dynamic-router-eip) | EIP | An example on how to use the Dynamic Router EIP component in Spring Boot
+| link:dynamic-router-eip/README.adoc[Dynamic Router Eip] (dynamic-router-eip) | EIP | Dynamic Router EIP component examples
| link:load-balancer-eip/README.adoc[Load Balancer Eip] (load-balancer-eip) | EIP | An example showing Load Balancer EIP with Camel and Spring Boot
diff --git a/dynamic-router-eip/README.adoc b/dynamic-router-eip/README.adoc
index c59586f..c6e8161 100644
--- a/dynamic-router-eip/README.adoc
+++ b/dynamic-router-eip/README.adoc
@@ -1,133 +1,152 @@
-== Spring Boot Example with the Dynamic Router EIP Component
-
-=== Example Description
-
-This example demonstrates how you might use the Dynamic Router EIP component in a Spring Boot application. There are eleven routing participants that subscribe to a "numbers" channel of the Dynamic Router. The message includes:
-
- 1. A Predicate that examines the Exchange to determine if an Exchange is appropriate for the participant
- 2. A destination URI where the Dynamic Router will send Exchanges that match a participant's Predicate
-
-.Routing Participants
-[cols="1,1,2"]
-|===
-|Name |Priority |Description
-
-|Tens
-|1
-|Accepts multiples of 10.
-
-|Nines
-|2
-|Accepts multiples of 9.
-
-|Eights
-|3
-|Accepts multiples of 8.
-
-|Sevens
-|4
-|Accepts multiples of 7.
-
-|Sixes
-|5
-|Accepts multiples of 6.
-
-|Fives
-|6
-|Accepts multiples of 5.
-
-|Fours
-|7
-|Accepts multiples of 4.
-
-|Threes
-|8
-|Accepts multiples of 3.
-
-|Even
-|9
-|Accepts even numbers.
-
-|Odd
-|100
-|Accepts odd numbers.
-
-|Prime
-|10
-|Accepts prime numbers.
-|===
-
-Subscriptions with a lower priority value are evaluated earlier than subscriptions with a higher priority value. For this reason, even though the number "10" is a multiple of 5, it will be consumed by the Tens participant.
-
-All participants inherit from the link:src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java[RoutingParticipant] class, so they use a `@Consume` annotation to consume from their registered destination URI. They add their received number to the link:src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java[ResultsService].
-
-The link:src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java[NumbersService] sends all numbers from one to one million through the Dynamic Router. After the messages have all been sent and processed, it instructs the ResultsService to print the results. It will look something like this:
-
-.Example Output
-[source,bash]
-----
- . ____ _ __ _ _
- /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
-( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
- \\/ ___)| |_)| | | | | || (_| | ) ) ) )
- ' |____| .__|_| |_|_| |_\__, | / / / /
- =========|_|==============|___/=/_/_/_/
- :: Spring Boot :: (v2.6.2)
-
-INFO 23130 --- [main] o.a.c.e.s.n.service.NumbersService : Subscribing participants
-INFO 23130 --- [main] o.a.c.e.s.n.service.NumbersService : Sending messages to the dynamic router
-INFO 23130 --- [main] o.a.c.e.s.n.service.NumbersService : Finished
-Dynamic Router Spring Boot Numbers Example Results:
- odd: 150077
- even: 114287
- sevens: 101588
- tens: 100000
- nines: 100000
- eights: 88889
- primes: 78494
- threes: 76191
- fives: 76190
- sixes: 57142
- fours: 57142
-Received count: 1000000 in 5709ms
-----
-
-=== Build
-
-You can build this example using:
-
- $ mvn package
-
-=== Run
-
-You can run this example using:
-
- $ mvn spring-boot:run
-
-If you want to run the Dynamic Router in "allMatch" receiver mode, run the example with an argument specifying that configuration property:
-
- $ mvn spring-boot:run -Dspring-boot.run.arguments="--camel.spring-boot.example.dynamic-router-eip.recipient-mode=allMatch"
-
-The results should show up in less than ten seconds, and the program will immediately terminate.
-
-=== Run Tests
-
-You can run the tests using:
-
- $mvn test
-
-There are some demos that run during the test phase. Currently, there are three:
-
- 1. AllRecipientsApplicationTest: the recipient mode is set to "allMatch" and 1,000,000 messages are sent through the router.
- 2. FirstRecipientApplicationTest: the recipient mode is set to "firstMatch" and 1,000,000 messages are sent through the router.
- 3. LessExpectedApplicationTest: the recipient mode is set to "firstMatch", and 10,000 messages are sent through the router. The difference, here, is that the expected count is set to only 1,000 messages, so the results are calculated and displayed before all 10,000 messages are routed.
-
-=== Help and contributions
-
-If you hit any problem using Camel or have some feedback, then please
-https://camel.apache.org/community/support/[let us know].
-
-We also love contributors, so please
-https://camel.apache.org/community/contributing/[get involved]
-
-The Camel riders!
+= Camel Dynamic Router EIP Component Spring Boot Examples
+
+This Camel Dynamic Router Spring Boot example module contains applications highlighting the two main use cases
+of the Dynamic Router EIP component.
+
+== Single JVM Example
+
+This example shows how you can route messages within a single application, or JVM.
+
+=== Build the Single JVM Example
+
+ mvn clean package -pl :camel-example-spring-boot-dynamic-router-eip-single
+
+=== Run the Single JVM Example
+
+ mvn spring-boot:run -pl :camel-example-spring-boot-dynamic-router-eip-single
+
+==== Access the Swagger UI
+
+Next, point your browser to http://localhost:8080/dynamic-router-example/swagger-ui.html. Here, you will find
+a single endpoint to interact with, called [.olive-background]#/generate#. Expand the section by clicking on
+the colored bar, and click on the `Try it out` button. There is only one field to fill in, and that is the number
+of messages that you want to send through the dynamic router. Enter a number, and then press the
+[.blue-background]#Execute# button. You will see a `Loading` animation while the messages are being processed.
+
+== Multiple JVM (Multimodule) Example
+
+This example shows how you can route messages between separate modules, where each runs in its own JVM.
+
+=== Build the Multimodule Example
+
+==== For systems with Docker
+
+ mvn clean package -f dynamic-router-eip/dynamic-router-eip-stack -Pdocker
+
+==== For systems with Podman
+
+ mvn clean package -f dynamic-router-eip/dynamic-router-eip-stack -Ppodman
+
+On a system with Podman, your output might resemble the following:
+
+[source,text]
+----
+Emulate Docker CLI using podman. Create /etc/containers/nodocker to quiet msg.
+>>>> Executing external compose provider "/usr/bin/docker-compose". Please refer to the documentation for details. <<<<
+
+[+] Running 6/6
+ ✔ Network docker_test-dynamic-router Created 0.0s
+ ✔ Container broker Healthy 0.1s
+ ✔ Container main_router_service Healthy 0.1s
+ ✔ Container odd_numbers_service Started 0.1s
+ ✔ Container even_numbers_service Started 0.1s
+ ✔ Container all_numbers_service Started 0.1s
+----
+
+=== Run the Multimodule Example
+
+When running the application stack for the Multimodule example, it should be the same, whether you have
+Docker or Podman installed. For Podman, please confirm that you have `podman compose` installed to ensure
+that you can emulate the Docker CLI when using Podman.
+
+ docker compose -f dynamic-router-eip/dynamic-router-eip-stack/project-resources/docker/docker-compose.yaml up -d
+
+==== Access the Swagger UI
+
+Next, point your browser to http://localhost:8082/main-router/swagger-ui.html. Here, you will find a few
+endpoints to interact with.
+
+===== Generate number messages
+
+The first endpoint is [.olive-background]#/generate#. Expand the section by clicking on the colored bar, and
+click on the `Try it out` button. There is only one field to fill in, and that is the number of messages that
+you want to send through the dynamic router. Enter a number, and then press the [.blue-background]#Execute#
+button. You will see a `Loading` animation while the messages are being generated. This will be very brief,
+since the message generation happens in its own thread, and everything at this point happens asynchronously.
+
+===== Retrieve number statistics
+
+From the moment messages begin to flow through the system, statistics are compiled. You can monitor the progress
+by using the next endpoint: [.blue-background]#/counts#. Expand the section by clicking on the colored bar, and
+click on the `Try it out` button. Click the [.blue-background]#Execute# button, and the current statistics will
+be displayed. These will continue to count upward until the current batch of messages has completed.
+
+A run of one million messages might yield output that resembles the following:
+
+[source,json]
+----
+{
+ "all": 1000000,
+ "odd": 500000,
+ "elapsed seconds": 19,
+ "even": 500000
+}
+----
+
+===== View subscriber information
+
+This endpoint allows you to see the routing participants that have subscribed for dynamic routing. Expand the
+endpoint for [.blue-background]#/list{channel}#. Clicking the `Try it out` button will allow you to enter the
+channel name. For this example app, you should enter `numbers` in that field, and then click the
+[.blue-background]#Execute# button. The returned text should look like this, although the formatting has been
+altered for better documentation clarity:
+
+[source,text]
+----
+[
+ PrioritizedFilterProcessor [
+ id: processAllNumbers,
+ priority: 5,
+ predicate: SpelExpression[
+ #{headers.command == 'processNumber' or headers.command == 'resetStats'}
+ ],
+ endpoint: kafka://numbers_all?groupInstanceId=numbers_all_consumer&headerDeserializer=#stringValueHeaderDeserializer
+ ],
+ PrioritizedFilterProcessor [
+ id: processEvenNumbers,
+ priority: 10,
+ predicate: SpelExpression[
+ #{(headers.command == 'processNumber' and headers.number matches '\d*[02468]') or
+ headers.command == 'resetStats'}
+ ],
+ endpoint: kafka://numbers_even?groupInstanceId=numbers_even_consumer&headerDeserializer=#stringValueHeaderDeserializer
+ ],
+ PrioritizedFilterProcessor [
+ id: processNumberStats,
+ priority: 10,
+ predicate: SpelExpression[
+ #{headers.command == 'stats'}
+ ],
+ endpoint: kafka://main_router?groupInstanceId=main_router_consumer&headerDeserializer=#stringValueHeaderDeserializer
+ ],
+ PrioritizedFilterProcessor [
+ id: processOddNumbers,
+ priority: 10,
+ predicate: SpelExpression[
+ #{(headers.command == 'processNumber' and headers.number matches '\d*[13579]') or
+ headers.command == 'resetStats'}
+ ],
+ endpoint: kafka://numbers_odd?groupInstanceId=numbers_odd_consumer&headerDeserializer=#stringValueHeaderDeserializer
+ ]
+]
+----
+
+== Help and Contributions
+
+If you hit any problem using Camel or have some feedback, then please
+https://camel.apache.org/community/support/[let us know].
+
+We also love contributors, so please
+https://camel.apache.org/community/contributing/[get involved]
+
+The Camel riders!
diff --git a/dynamic-router-eip/dynamic-router-eip-single/pom.xml b/dynamic-router-eip/dynamic-router-eip-single/pom.xml
new file mode 100644
index 0000000..5f4a5d0
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-single/pom.xml
@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-single</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Single JVM Example</name>
+ <description>An example on how to use the Dynamic Router EIP component in a single Spring Boot application</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.springdoc</groupId>
+ <artifactId>springdoc-openapi-starter-webflux-ui</artifactId>
+ <version>${springdoc-version}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/Application.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/Application.java
similarity index 93%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/Application.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/Application.java
index 64752fa..efc7fa5 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/Application.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/Application.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.example.springboot;
+package org.apache.camel.example.springboot.numbers;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java
similarity index 73%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java
index a7403e6..b81ead5 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/DynamicRouterComponentConfig.java
@@ -16,22 +16,24 @@
*/
package org.apache.camel.example.springboot.numbers.config;
+import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.Predicate;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.dynamicrouter.DynamicRouterConfiguration;
-import org.apache.camel.component.dynamicrouter.DynamicRouterConstants;
import org.apache.camel.example.springboot.numbers.participants.PredicateConstants;
import org.apache.camel.example.springboot.numbers.participants.RoutingParticipant;
import org.apache.camel.example.springboot.numbers.service.ResultsService;
+import org.apache.camel.impl.engine.DefaultExecutorServiceManager;
+import org.apache.camel.support.DefaultThreadPoolFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
-import static org.apache.camel.component.dynamicrouter.DynamicRouterConstants.COMPONENT_SCHEME;
-import static org.apache.camel.component.dynamicrouter.DynamicRouterConstants.MODE_FIRST_MATCH;
+import static org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.COMPONENT_SCHEME_ROUTING;
import static org.apache.camel.example.springboot.numbers.participants.PredicateConstants.PREDICATE_EIGHTS;
import static org.apache.camel.example.springboot.numbers.participants.PredicateConstants.PREDICATE_EVEN;
import static org.apache.camel.example.springboot.numbers.participants.PredicateConstants.PREDICATE_FIVES;
@@ -46,7 +48,7 @@ import static org.apache.camel.example.springboot.numbers.participants.Predicate
/**
* This configuration ingests the config properties in the application.yml file.
- * Sets up the Camel route that feeds the Dynamic Router. Also creates all of the
+ * Sets up the Camel route that feeds the Dynamic Router. Also creates all
* routing participants by using the various predicates defined in the
* {@link PredicateConstants} class.
*/
@@ -59,13 +61,35 @@ public class DynamicRouterComponentConfig {
*/
private final ExampleConfig exampleConfig;
+ /**
+ * The Camel context.
+ */
+ private final CamelContext camelContext;
+
/**
* Create this config with the config properties object.
*
* @param exampleConfig the config properties object
*/
- public DynamicRouterComponentConfig(final ExampleConfig exampleConfig) {
+ public DynamicRouterComponentConfig(final ExampleConfig exampleConfig, CamelContext camelContext) {
this.exampleConfig = exampleConfig;
+ this.camelContext = camelContext;
+ }
+
+ private static final Function<ExampleConfig, String> createCommandUri = cfg ->
+ "%s:%s?recipientMode=%s&executorService=%s".formatted(
+ COMPONENT_SCHEME_ROUTING,
+ cfg.getRoutingChannel(),
+ cfg.getRecipientMode(),
+ "customPool");
+
+ @Bean
+ public ExecutorService customPool() {
+ DefaultThreadPoolFactory threadPoolFactory = new DefaultThreadPoolFactory();
+ threadPoolFactory.setCamelContext(camelContext);
+ DefaultExecutorServiceManager executorServiceManager = new DefaultExecutorServiceManager(camelContext);
+ executorServiceManager.setThreadPoolFactory(threadPoolFactory);
+ return executorServiceManager.newCachedThreadPool(this, "DynamicRouterSpringBootThreadFactory");
}
/**
@@ -77,37 +101,68 @@ public class DynamicRouterComponentConfig {
return new RouteBuilder() {
@Override
public void configure() {
- from(exampleConfig.getStartUri())
- .to(COMPONENT_SCHEME + ":" + exampleConfig.getRoutingChannel() + "?recipientMode=" + exampleConfig.getRecipientMode());
+ from(exampleConfig.getStartUri()).to(createCommandUri.apply(exampleConfig));
}
};
}
- /**
- * Create a {@link CountDownLatch} so that we can wait on it for the total
- * expected number of messages to be received. Depending on the configured
- * value of {@link ExampleConfig#getRecipientMode()}, this latch will be
- * created with {@link ExampleConfig#getExpectedAllMatchMessageCount()} or
- * {@link ExampleConfig#getExpectedFirstMatchMessageCount()}.
- *
- * @see DynamicRouterConstants#MODE_ALL_MATCH
- * @see DynamicRouterConstants#MODE_FIRST_MATCH
- * @see DynamicRouterConfiguration#getRecipientMode()
- * @see ExampleConfig#getRecipientMode()
- * @see ExampleConfig#getExpectedAllMatchMessageCount()
- * @see ExampleConfig#getExpectedFirstMatchMessageCount()
- *
- * @return a countdown latch set to the number of expected messages
- */
@Bean
- CountDownLatch countDownLatch() {
- return new CountDownLatch(MODE_FIRST_MATCH.equals(exampleConfig.getRecipientMode()) ?
- exampleConfig.getExpectedFirstMatchMessageCount() :
- exampleConfig.getExpectedAllMatchMessageCount());
+ Predicate predicateTens() {
+ return PREDICATE_TENS;
+ }
+
+ @Bean
+ Predicate predicateNines() {
+ return PREDICATE_NINES;
+ }
+
+ @Bean
+ Predicate predicateEights() {
+ return PREDICATE_EIGHTS;
+ }
+
+ @Bean
+ Predicate predicateSevens() {
+ return PREDICATE_SEVENS;
+ }
+
+ @Bean
+ Predicate predicateSixes() {
+ return PREDICATE_SIXES;
+ }
+
+ @Bean
+ Predicate predicateFives() {
+ return PREDICATE_FIVES;
+ }
+
+ @Bean
+ Predicate predicateFours() {
+ return PREDICATE_FOURS;
+ }
+
+ @Bean
+ Predicate predicateThrees() {
+ return PREDICATE_THREES;
+ }
+
+ @Bean
+ Predicate predicateEven() {
+ return PREDICATE_EVEN;
+ }
+
+ @Bean
+ Predicate predicateOdd() {
+ return PREDICATE_ODD;
+ }
+
+ @Bean
+ Predicate predicatePrimes() {
+ return PREDICATE_PRIMES;
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 10.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -116,11 +171,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant tensRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "tens", 1, PREDICATE_TENS, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "tens", 1, "predicateTens", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 9.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -129,11 +184,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant ninesRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "nines", 2, PREDICATE_NINES, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "nines", 2, "predicateNines", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 8.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -143,11 +198,11 @@ public class DynamicRouterComponentConfig {
@Bean
RoutingParticipant eightsRoutingParticipant(
final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "eights", 3, PREDICATE_EIGHTS, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "eights", 3, "predicateEights", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 7.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -157,11 +212,11 @@ public class DynamicRouterComponentConfig {
@Bean
RoutingParticipant sevensRoutingParticipant(
final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "sevens", 4, PREDICATE_SEVENS, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "sevens", 4, "predicateSevens", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 6.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -170,11 +225,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant sixesRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "sixes", 5, PREDICATE_SIXES, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "sixes", 5, "predicateSixes", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 5.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -183,11 +238,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant fivesRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "fives", 6, PREDICATE_FIVES, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "fives", 6, "predicateFives", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 4.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -196,11 +251,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant foursRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "fours", 7, PREDICATE_FOURS, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "fours", 7, "predicateFours", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a multiple of 3.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -210,11 +265,11 @@ public class DynamicRouterComponentConfig {
@Bean
RoutingParticipant threesRoutingParticipant(
final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "threes", 8, PREDICATE_THREES, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "threes", 8, "predicateThrees", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* an even number.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -223,11 +278,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant evensRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "even", 9, PREDICATE_EVEN, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "even", 9, "predicateEven", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* an odd number.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -236,11 +291,11 @@ public class DynamicRouterComponentConfig {
*/
@Bean
RoutingParticipant oddsRoutingParticipant(final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "odd", 100, PREDICATE_ODD, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "odd", 100, "predicateOdd", subscriberTemplate, resultsService);
}
/**
- * Create a {@link RoutingParticipant} that handles messages where the body is comprised of a number that is
+ * Create a {@link RoutingParticipant} that handles messages where the body comprises a number that is
* a prime.
*
* @param subscriberTemplate the {@link ConsumerTemplate} for subscribing to messages that match
@@ -250,6 +305,6 @@ public class DynamicRouterComponentConfig {
@Bean
RoutingParticipant primesRoutingParticipant(
final ProducerTemplate subscriberTemplate, final ResultsService resultsService) {
- return new RoutingParticipant(exampleConfig, "primes", 10, PREDICATE_PRIMES, subscriberTemplate, resultsService);
+ return new RoutingParticipant(exampleConfig, "primes", 10, "predicatePrimes", subscriberTemplate, resultsService);
}
}
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java
similarity index 59%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java
index 187925f..ed1fc38 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/config/ExampleConfig.java
@@ -18,8 +18,6 @@ package org.apache.camel.example.springboot.numbers.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
-import java.util.concurrent.CountDownLatch;
-
/**
* The config properties object from the application.yml file.
*/
@@ -57,24 +55,6 @@ public class ExampleConfig {
*/
private int sendMessageCount;
- /**
- * If the recipient mode is "firstMatch", then this is the number of
- * messages that are expected to be received. This value is also used
- * to create a {@link CountDownLatch} so that all messages can be
- * received before printing out messaging statistics.
- */
- private int expectedFirstMatchMessageCount;
-
- /**
- * If the recipient mode is "allMatch", then this is the number of
- * messages that are expected to be received. There is probably not an
- * easy way to calculate this number, so the number needs to be known
- * for a given {@link #sendMessageCount} value. This value is also used
- * to create a {@link CountDownLatch} so that all messages can be
- * received before printing out messaging statistics.
- */
- private int expectedAllMatchMessageCount;
-
/**
* The URI where messages will be sent to as the starting point for the
* route that feeds the dynamic router.
@@ -141,34 +121,4 @@ public class ExampleConfig {
public void setSendMessageCount(int sendMessageCount) {
this.sendMessageCount = sendMessageCount;
}
-
- /**
- * If the recipient mode is "firstMatch", then this is the number of
- * messages that are expected to be received. This value is also used
- * to create a {@link CountDownLatch} so that all messages can be
- * received before printing out messaging statistics.
- */
- public int getExpectedFirstMatchMessageCount() {
- return expectedFirstMatchMessageCount;
- }
-
- public void setExpectedFirstMatchMessageCount(int expectedFirstMatchMessageCount) {
- this.expectedFirstMatchMessageCount = expectedFirstMatchMessageCount;
- }
-
- /**
- * If the recipient mode is "allMatch", then this is the number of
- * messages that are expected to be received. There is probably not an
- * easy way to calculate this number, so the number needs to be known
- * for a given {@link #sendMessageCount} value. This value is also used
- * to create a {@link CountDownLatch} so that all messages can be
- * received before printing out messaging statistics.
- */
- public int getExpectedAllMatchMessageCount() {
- return expectedAllMatchMessageCount;
- }
-
- public void setExpectedAllMatchMessageCount(int expectedAllMatchMessageCount) {
- this.expectedAllMatchMessageCount = expectedAllMatchMessageCount;
- }
}
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java
similarity index 62%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java
index 7f441d6..9db8f92 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/PredicateConstants.java
@@ -19,70 +19,85 @@ package org.apache.camel.example.springboot.numbers.participants;
import org.apache.camel.Exchange;
import org.apache.camel.Predicate;
-import java.util.function.BiFunction;
+import java.util.function.BiPredicate;
+import java.util.function.Function;
+import java.util.function.IntFunction;
+import java.util.function.ToIntFunction;
/**
* Provides various {@link Predicate}s that routing participants can send to the
* dynamic router as rules to determine exchange suitability.
*/
-public abstract class PredicateConstants {
+public final class PredicateConstants {
+
+ private PredicateConstants() {
+ throw new UnsupportedOperationException("Utility class should not be instantiated");
+ }
/**
* Gets the message body as an integer and determines if the number is evenly
* divided by the supplied integer.
*/
- public static final BiFunction<Exchange, Integer, Boolean> noRemainder = (e, m) ->
- e.getIn().getBody(Integer.class) % m == 0;
+ public static final BiPredicate<Integer, Integer> noRemainder = (n, m) -> n % m == 0;
+
+ /**
+ * Extracts the number from the exchange header.
+ */
+ public static final ToIntFunction<Exchange> extractValue = e ->
+ e.getMessage().getHeader("number", Integer.class);
+
+ public static final IntFunction<Predicate> noRemainderCurried = m ->
+ e -> noRemainder.test(extractValue.applyAsInt(e), m);
/**
* Determines if the message body is a number that is even.
*/
- public static final Predicate PREDICATE_EVEN = e -> noRemainder.apply(e, 2);
+ public static final Predicate PREDICATE_EVEN = e -> noRemainderCurried.apply(2).matches(e);
/**
* Determines if the message body is a number that is odd.
*/
- public static final Predicate PREDICATE_ODD = e -> e.getIn().getBody(Integer.class) % 2 != 0;
+ public static final Predicate PREDICATE_ODD = e -> !noRemainderCurried.apply(2).matches(e);
/**
* Determines if the message body is a number that is a multiple of 3.
*/
- public static final Predicate PREDICATE_THREES = e -> noRemainder.apply(e, 3);
+ public static final Predicate PREDICATE_THREES = e -> noRemainderCurried.apply(3).matches(e);
/**
* Determines if the message body is a number that is a multiple of 4.
*/
- public static final Predicate PREDICATE_FOURS = e -> noRemainder.apply(e, 4);
+ public static final Predicate PREDICATE_FOURS = e -> noRemainderCurried.apply(4).matches(e);
/**
* Determines if the message body is a number that is a multiple of 5.
*/
- public static final Predicate PREDICATE_FIVES = e -> noRemainder.apply(e, 5);
+ public static final Predicate PREDICATE_FIVES = e -> noRemainderCurried.apply(5).matches(e);
/**
* Determines if the message body is a number that is a multiple of 6.
*/
- public static final Predicate PREDICATE_SIXES = e -> noRemainder.apply(e, 6);
+ public static final Predicate PREDICATE_SIXES = e -> noRemainderCurried.apply(6).matches(e);
/**
* Determines if the message body is a number that is a multiple of 7.
*/
- public static final Predicate PREDICATE_SEVENS = e -> noRemainder.apply(e, 7);
+ public static final Predicate PREDICATE_SEVENS = e -> noRemainderCurried.apply(7).matches(e);
/**
* Determines if the message body is a number that is a multiple of 8.
*/
- public static final Predicate PREDICATE_EIGHTS = e -> noRemainder.apply(e, 8);
+ public static final Predicate PREDICATE_EIGHTS = e -> noRemainderCurried.apply(8).matches(e);
/**
* Determines if the message body is a number that is a multiple of 9.
*/
- public static final Predicate PREDICATE_NINES = e -> noRemainder.apply(e, 9);
+ public static final Predicate PREDICATE_NINES = e -> noRemainderCurried.apply(9).matches(e);
/**
* Determines if the message body is a number that is a multiple of 10.
*/
- public static final Predicate PREDICATE_TENS = e -> noRemainder.apply(e, 10);
+ public static final Predicate PREDICATE_TENS = e -> noRemainderCurried.apply(10).matches(e);
/**
* If this predicate is prioritized with a higher number than {@link #PREDICATE_SEVENS}
@@ -90,20 +105,24 @@ public abstract class PredicateConstants {
* and 2 in the accumulated list of prime numbers.
*/
public static final Predicate PREDICATE_PRIMES = e -> {
- int n = e.getIn().getBody(Integer.class);
- // 2 is the first prime number
- if (n <= 2) {
- return n == 2;
+ int n = extractValue.applyAsInt(e);
+ if (n <= 0) {
+ return false;
+ }
+ // handle cases for 1, 2 and 3
+ if (n <= 3) {
+ return n > 1;
}
- // no other even numbers are prime
- if (n % 2 == 0) {
+ // check if number is divisible by 2 or 3
+ if (noRemainder.test(n, 2) || noRemainder.test(n, 3)) {
return false;
}
- // only some odd numbers might be prime
- int max = (int) Math.sqrt(n) + 1;
- for (int i = 3; i < max; i += 2) {
- if (n % i == 0)
+ int i = 5;
+ while (i * i <= n) {
+ if (noRemainder.test(n, i) || noRemainder.test(n, (i + 2))) {
return false;
+ }
+ i += 6;
}
return true;
};
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java
similarity index 85%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java
index 62d7b25..62c9025 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/participants/RoutingParticipant.java
@@ -19,8 +19,7 @@ package org.apache.camel.example.springboot.numbers.participants;
import org.apache.camel.Consume;
import org.apache.camel.Predicate;
import org.apache.camel.ProducerTemplate;
-import org.apache.camel.component.dynamicrouter.DynamicRouterControlMessage;
-import org.apache.camel.component.dynamicrouter.DynamicRouterControlMessage.SubscribeMessageBuilder;
+import org.apache.camel.component.dynamicrouter.control.DynamicRouterControlMessage;
import org.apache.camel.example.springboot.numbers.config.ExampleConfig;
import org.apache.camel.example.springboot.numbers.service.ResultsService;
@@ -64,10 +63,10 @@ public class RoutingParticipant {
protected final int priority;
/**
- * The {@link Predicate} by which exchanges are evaluated for suitability for
+ * The name of the {@link Predicate} by which exchanges are evaluated for suitability for
* a routing participant.
*/
- protected final Predicate predicate;
+ protected final String predicate;
/**
* The URI that a participant implementation will listen on for messages
@@ -91,7 +90,7 @@ public class RoutingParticipant {
final ExampleConfig config,
final String bin,
final int priority,
- final Predicate predicate,
+ final String predicate,
final ProducerTemplate subscriberTemplate,
final ResultsService resultsService) {
this.bin = bin;
@@ -152,10 +151,10 @@ public class RoutingParticipant {
}
/**
- * The implementation's rule as a {@link Predicate}.
- * @return the {@link Predicate} rule
+ * The name of the implementation's rule as a {@link Predicate}.
+ * @return the {@link Predicate} rule name
*/
- protected Predicate getPredicate() {
+ protected String getPredicate() {
return predicate;
}
@@ -166,12 +165,12 @@ public class RoutingParticipant {
* @return the {@link DynamicRouterControlMessage}
*/
protected DynamicRouterControlMessage createSubscribeMessage() {
- return new SubscribeMessageBuilder()
- .id(getBin())
- .channel(routingChannel)
+ return DynamicRouterControlMessage.Builder.newBuilder()
+ .subscriptionId(getBin())
+ .subscribeChannel(routingChannel)
.priority(getPriority())
- .endpointUri(getConsumeUri())
- .predicate(getPredicate())
+ .destinationUri(getConsumeUri())
+ .predicateBean(getPredicate())
.build();
}
}
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java
similarity index 77%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java
index e23c81b..4fed466 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/NumbersService.java
@@ -23,18 +23,15 @@ import org.apache.camel.example.springboot.numbers.participants.RoutingParticipa
import org.apache.camel.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.boot.context.event.ApplicationReadyEvent;
-import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
-import java.text.NumberFormat;
import java.util.List;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-import java.util.stream.IntStream;
-import static java.util.concurrent.TimeUnit.*;
+import static java.util.concurrent.TimeUnit.MINUTES;
/**
* Create numbers and send them to the dynamic router so that they can be
@@ -46,8 +43,6 @@ public class NumbersService {
private static final Logger LOG = LoggerFactory.getLogger(NumbersService.class);
- private final NumberFormat numberFormat = NumberFormat.getIntegerInstance();
-
/**
* The URI to send the messages to. This URI feeds the dynamic router in a
* Camel route.
@@ -76,7 +71,7 @@ public class NumbersService {
* The {@link CountDownLatch} to wait for the total expected number of
* messages to be received.
*/
- final CountDownLatch countDownLatch;
+ private CountDownLatch countDownLatch;
/**
* The number of messages to generate.
@@ -90,19 +85,16 @@ public class NumbersService {
* @param participants the dynamic router participants
* @param start the producer template to send messages to the start endpoint
* @param resultsService the service that compiles routing results
- * @param countDownLatch the latch to wait for all messages to be received
*/
public NumbersService(
final ExampleConfig config,
final List<RoutingParticipant> participants,
final ProducerTemplate start,
- final ResultsService resultsService,
- final CountDownLatch countDownLatch) {
+ final ResultsService resultsService) {
this.startUri = config.getStartUri();
this.participants = participants;
this.start = start;
this.resultsService = resultsService;
- this.countDownLatch = countDownLatch;
this.numberOfMessages = config.getSendMessageCount();
}
@@ -111,24 +103,33 @@ public class NumbersService {
* participant to subscribe, and then send the messages. Afterward,
* display the results and exit the app.
*/
- @EventListener(ApplicationReadyEvent.class)
- public void start() throws InterruptedException {
+ public String start(int limit) throws InterruptedException {
+ resultsService.resetStatistics();
LOG.info("Subscribing {} participants", participants.size());
participants.forEach(RoutingParticipant::subscribe);
+ countDownLatch = new CountDownLatch(1);
final StopWatch watch = new StopWatch();
- LOG.info("Sending {} messages to the dynamic router: {}", numberFormat.format(numberOfMessages), getStartUri());
- CompletableFuture.runAsync(this::sendMessages, Executors.newSingleThreadExecutor());
+ LOG.info("Sending {} messages to the dynamic router: {}", limit, getStartUri());
+ Mono<Integer> msgFlux = sendMessages(limit);
+ msgFlux.subscribe();
if (!countDownLatch.await(1, MINUTES)) {
LOG.warn("Statistics may be inaccurate, since the operation timed out");
}
- LOG.info(resultsService.getStatistics(watch));
+ return resultsService.getStatistics(watch, limit);
}
/**
* Sends the messages to the starting endpoint of the route.
+ *
+ * @return an empty Mono
*/
- public void sendMessages() {
- IntStream.rangeClosed(1, numberOfMessages).forEach(start::sendBody);
+ public Mono<Integer> sendMessages(int limit) {
+ return Flux.range(1, limit)
+ .parallel()
+ .runOn(Schedulers.boundedElastic())
+ .doOnNext(n -> start.sendBodyAndHeader(n, "number", n))
+ .reduce((a, b) -> b)
+ .doFinally(signal -> countDownLatch.countDown());
}
/**
diff --git a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java
similarity index 72%
rename from dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java
index c291ef2..292994d 100644
--- a/dynamic-router-eip/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/service/ResultsService.java
@@ -16,17 +16,14 @@
*/
package org.apache.camel.example.springboot.numbers.service;
-import org.apache.camel.example.springboot.numbers.config.ExampleConfig;
import org.apache.camel.util.StopWatch;
import org.springframework.stereotype.Service;
import java.text.NumberFormat;
-import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ConcurrentSkipListSet;
/**
* Holds processing results so that they can be displayed, etc.
@@ -39,26 +36,13 @@ public class ResultsService {
/**
* A map of the "bin" or "category" to the list of values in that bin.
*/
- private final ConcurrentHashMap<String, Collection<Integer>> results;
-
- /**
- * The {@link CountDownLatch} to wait for all messages to be received.
- */
- private final CountDownLatch countDownLatch;
-
- /**
- * The number of sent messages.
- */
- private final int numberSent;
+ private final ConcurrentSkipListMap<String, ConcurrentSkipListSet<Integer>> results;
/**
* Create the service and initialize the results map.
*/
- public ResultsService(final CountDownLatch countDownLatch,
- final ExampleConfig config) {
- this.results = new ConcurrentHashMap<>();
- this.countDownLatch = countDownLatch;
- this.numberSent = config.getSendMessageCount();
+ public ResultsService() {
+ this.results = new ConcurrentSkipListMap<>();
}
/**
@@ -68,8 +52,7 @@ public class ResultsService {
* @param value the value to add to the bin
*/
public void addResult(final String key, int value) {
- results.computeIfAbsent(key, v -> Collections.synchronizedCollection(new ArrayList<>())).add(value);
- countDownLatch.countDown();
+ results.computeIfAbsent(key, v -> new ConcurrentSkipListSet<>()).add(value);
}
/**
@@ -77,36 +60,39 @@ public class ResultsService {
*
* @return the results map
*/
- public Map<String, Collection<Integer>> getResults() {
+ public Map<String, ConcurrentSkipListSet<Integer>> getResults() {
return results;
}
+ public void resetStatistics() {
+ results.clear();
+ }
+
/**
* Get a message that contains the statistics of the messaging.
*
* @param watch a {@link StopWatch} that was started at the beginning of messaging
* @return a message that contains the statistics of the messaging
*/
- public String getStatistics(final StopWatch watch) {
+ public String getStatistics(final StopWatch watch, int numberSent) {
final long taken = watch.taken();
final int totalCount = getResults().values()
.stream()
.mapToInt(Collection::size)
.sum();
- final int numberLength = numberFormat.format(totalCount)
- .length();
+ final int numberLength = numberFormat.format(totalCount).length();
StringBuilder statistics = new StringBuilder("Finished in ")
.append(taken).append("ms")
.append("\nDynamic Router Spring Boot Numbers Example Results:\n");
getResults().entrySet().stream()
.sorted((o1, o2) -> o2.getValue().size() - o1.getValue().size())
- .map(e -> String.format("%7s: %" + numberLength + "s [%3d%% routed, %3d%% sent]",
+ .map(e -> String.format("%7s: %" + numberLength + "s [%3d%% routed, %3d%% sent]", // NOSONAR
e.getKey(), numberFormat.format(e.getValue().size()),
e.getValue().size() * 100 / totalCount,
e.getValue().size() * 100 / numberSent))
.forEach(s -> statistics.append("\n\t").append(s));
statistics.append("\n\n\t")
- .append(String.format("%7s: %" + numberLength + "s [%3d%% routed, %3d%% sent]",
+ .append(String.format("%7s: %" + numberLength + "s [%3d%% routed, %3d%% sent]", // NOSONAR
"total", numberFormat.format(totalCount),
totalCount * 100 / totalCount,
totalCount * 100 / numberSent));
diff --git a/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/web/NumbersController.java b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/web/NumbersController.java
new file mode 100644
index 0000000..8cab95a
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/java/org/apache/camel/example/springboot/numbers/web/NumbersController.java
@@ -0,0 +1,22 @@
+package org.apache.camel.example.springboot.numbers.web;
+
+import org.apache.camel.example.springboot.numbers.service.NumbersService;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+@RestController
+public class NumbersController {
+
+ private final NumbersService numbersService;
+
+ public NumbersController(NumbersService numbersService) {
+ this.numbersService = numbersService;
+ }
+
+ @PutMapping(path = "/generate")
+ public Mono<String> generate(@RequestParam("limit") int limit) throws InterruptedException {
+ return Mono.just(numbersService.start(limit));
+ }
+}
diff --git a/dynamic-router-eip/src/main/resources/application.yml b/dynamic-router-eip/dynamic-router-eip-single/src/main/resources/application.yml
similarity index 81%
rename from dynamic-router-eip/src/main/resources/application.yml
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/resources/application.yml
index bf7bc21..4986e6e 100644
--- a/dynamic-router-eip/src/main/resources/application.yml
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/resources/application.yml
@@ -19,23 +19,18 @@ camel:
name: CamelSpringBootDynamicRouterExample
jmx-enabled: false
shutdown-timeout: 30
+ endpoint-bridge-error-handler: true
spring-boot:
example:
dynamic-router-eip:
routing-channel: numbers
- subscribe-uri: dynamic-router:control
+ subscribe-uri: dynamic-router-control:subscribe
receiver-scheme: direct
start-uri: direct:start
recipient-mode: firstMatch
- send-message-count: 1000000
- expected-first-match-message-count: 1000000
- expected-all-match-message-count: 2507465
component:
dynamic-router:
lazy-start-producer: true
- bridge-error-handler: true
- cloud:
- enabled: false
logging:
level:
root: WARN
@@ -43,3 +38,9 @@ logging:
spring:
profiles:
active: default
+ application:
+ name: CamelSpringBootDynamicRouterExample
+ webflux:
+ base-path: "/dynamic-router-example"
+server:
+ port: 8080
diff --git a/dynamic-router-eip/src/main/resources/logback.xml b/dynamic-router-eip/dynamic-router-eip-single/src/main/resources/logback.xml
similarity index 79%
rename from dynamic-router-eip/src/main/resources/logback.xml
rename to dynamic-router-eip/dynamic-router-eip-single/src/main/resources/logback.xml
index 3502b16..ab86d43 100644
--- a/dynamic-router-eip/src/main/resources/logback.xml
+++ b/dynamic-router-eip/dynamic-router-eip-single/src/main/resources/logback.xml
@@ -7,6 +7,6 @@
<root level="error">
<appender-ref ref="STDOUT"/>
</root>
- <logger name="org.apache.camel.example.springboot.Application" level="warn"/>
+ <logger name="org.apache.camel.example.springboot.numbers.Application" level="warn"/>
<logger name="org.apache.camel.example.springboot" level="info"/>
</configuration>
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/pom.xml
new file mode 100644
index 0000000..fc65dbb
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/pom.xml
@@ -0,0 +1,41 @@
+<!--
+ ~ 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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>all-numbers-service</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule :: All Numbers Service</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>numbers-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/Application.java b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/Application.java
new file mode 100644
index 0000000..973d7f5
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/Application.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.example.springboot.numbers.all;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@EnableScheduling
+@SpringBootApplication(scanBasePackages = "org.apache.camel.example.springboot.numbers")
+public class Application {
+
+ /**
+ * Main method to start the application. Please make us proud.
+ */
+ public static void main(String[] args) {
+ SpringApplication.run(Application.class, args);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/service/ProcessAllNumbersRoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/service/ProcessAllNumbersRoutingParticipant.java
new file mode 100644
index 0000000..11c02a0
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/all/service/ProcessAllNumbersRoutingParticipant.java
@@ -0,0 +1,43 @@
+/*
+ * 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.example.springboot.numbers.all.service;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.common.service.ProcessNumbersRoutingParticipant;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ProcessAllNumbersRoutingParticipant extends ProcessNumbersRoutingParticipant {
+
+ public ProcessAllNumbersRoutingParticipant(
+ @Value("${number-generator.subscribe-uri}") String subscribeUri,
+ @Value("${number-generator.routing-channel}") String routingChannel,
+ @Value("${number-generator.predicate}") String predicate,
+ @Value("${number-generator.expression-language}") String expressionLanguage,
+ @Value("${number-generator.subscription-priority}") int subscriptionPriority,
+ @Value("${number-generator.consume-uri}") String consumeUri,
+ @Value("${number-generator.command-uri}") String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ super("all", "processAllNumbers", subscribeUri, routingChannel, subscriptionPriority,
+ predicate, expressionLanguage, consumeUri, commandUri, producerTemplate, camelContext);
+
+ }
+}
diff --git a/dynamic-router-eip/src/test/resources/application-test-all.yml b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/resources/application.yaml
similarity index 50%
rename from dynamic-router-eip/src/test/resources/application-test-all.yml
rename to dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/resources/application.yaml
index f01b644..b4766ab 100644
--- a/dynamic-router-eip/src/test/resources/application-test-all.yml
+++ b/dynamic-router-eip/dynamic-router-eip-stack/all-numbers-service/src/main/resources/application.yaml
@@ -1,45 +1,22 @@
-#
-# 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.
-#
-camel:
- springboot:
- name: CamelSpringBootDynamicRouterExample
- jmx-enabled: false
- shutdown-timeout: 30
- spring-boot:
- example:
- dynamic-router-eip:
- routing-channel: numbers
- subscribe-uri: dynamic-router:control
- receiver-scheme: direct
- start-uri: direct:start
- recipient-mode: allMatch
- send-message-count: 1000000
- expected-first-match-message-count: 1000000
- expected-all-match-message-count: 2507465
- component:
- dynamic-router:
- lazy-start-producer: true
- bridge-error-handler: true
- cloud:
- enabled: false
-logging:
- level:
- root: WARN
- org.apache.camel.example.springboot: INFO
-spring:
- main:
- banner-mode: off
+#
+# 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.
+#
+app:
+ name: "all-numbers"
+ port: 8911
+ id: "numbers_all"
+ predicate: "{headers.command == 'processNumber' or headers.command == 'resetStats'}"
+ priority: 5
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/pom.xml
new file mode 100644
index 0000000..40cb246
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/pom.xml
@@ -0,0 +1,41 @@
+<!--
+ ~ 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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>even-numbers-service</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule :: Even Numbers Service</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>numbers-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/Application.java b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/Application.java
new file mode 100644
index 0000000..201c4f8
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/Application.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.example.springboot.numbers.even;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "org.apache.camel.example.springboot.numbers")
+public class Application {
+
+ /**
+ * Main method to start the application. Please make us proud.
+ */
+ public static void main(String[] args) {
+ SpringApplication.run(Application.class, args);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/service/ProcessEvenNumbersRoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/service/ProcessEvenNumbersRoutingParticipant.java
new file mode 100644
index 0000000..e5e0712
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/even/service/ProcessEvenNumbersRoutingParticipant.java
@@ -0,0 +1,42 @@
+/*
+ * 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.example.springboot.numbers.even.service;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.common.service.ProcessNumbersRoutingParticipant;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ProcessEvenNumbersRoutingParticipant extends ProcessNumbersRoutingParticipant {
+
+ public ProcessEvenNumbersRoutingParticipant(
+ @Value("${number-generator.subscribe-uri}") String subscribeUri,
+ @Value("${number-generator.routing-channel}") String routingChannel,
+ @Value("${number-generator.predicate}") String predicate,
+ @Value("${number-generator.expression-language}") String expressionLanguage,
+ @Value("${number-generator.subscription-priority}") int subscriptionPriority,
+ @Value("${number-generator.consume-uri}") String consumeUri,
+ @Value("${number-generator.command-uri}") String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ super("even", "processEvenNumbers", subscribeUri, routingChannel, subscriptionPriority,
+ predicate, expressionLanguage, consumeUri, commandUri, producerTemplate, camelContext);
+ }
+}
diff --git a/dynamic-router-eip/src/test/resources/application-test-first.yml b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/resources/application.yaml
similarity index 50%
rename from dynamic-router-eip/src/test/resources/application-test-first.yml
rename to dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/resources/application.yaml
index ae35410..9074dc4 100644
--- a/dynamic-router-eip/src/test/resources/application-test-first.yml
+++ b/dynamic-router-eip/dynamic-router-eip-stack/even-numbers-service/src/main/resources/application.yaml
@@ -1,45 +1,22 @@
-#
-# 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.
-#
-camel:
- springboot:
- name: CamelSpringBootDynamicRouterExample
- jmx-enabled: false
- shutdown-timeout: 30
- spring-boot:
- example:
- dynamic-router-eip:
- routing-channel: numbers
- subscribe-uri: dynamic-router:control
- receiver-scheme: direct
- start-uri: direct:start
- recipient-mode: firstMatch
- send-message-count: 1000000
- expected-first-match-message-count: 1000000
- expected-all-match-message-count: 2507465
- component:
- dynamic-router:
- lazy-start-producer: true
- bridge-error-handler: true
- cloud:
- enabled: false
-logging:
- level:
- root: WARN
- org.apache.camel.example.springboot: INFO
-spring:
- main:
- banner-mode: off
+#
+# 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.
+#
+app:
+ name: "even-numbers"
+ port: 8902
+ id: "numbers_even"
+ predicate: "{(headers.command == 'processNumber' and headers.number matches '\\d*[02468]') or headers.command == 'resetStats'}"
+ priority: 10
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/main-router/pom.xml
new file mode 100644
index 0000000..51c5db7
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/pom.xml
@@ -0,0 +1,68 @@
+<!--
+ ~ 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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>main-router</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule :: Main Router Service</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>numbers-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springdoc</groupId>
+ <artifactId>springdoc-openapi-starter-webflux-ui</artifactId>
+ <version>${springdoc-version}</version>
+ </dependency>
+ </dependencies>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <version>${spring-boot-version}</version>
+ <executions>
+ <!--
+ This can be removed when AOT processing is
+ compatible with spring statemachine
+ -->
+ <execution>
+ <id>process-aot</id>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/Application.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/Application.java
new file mode 100644
index 0000000..cdcef72
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/Application.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.example.springboot.numbers.mainrouter;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "org.apache.camel.example.springboot.numbers")
+public class Application {
+
+ /**
+ * Main method to start the application. Please make us proud.
+ */
+ public static void main(String[] args) {
+ SpringApplication.run(Application.class, args);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/DynamicRouterComponentConfig.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/DynamicRouterComponentConfig.java
new file mode 100644
index 0000000..403080e
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/DynamicRouterComponentConfig.java
@@ -0,0 +1,39 @@
+/*
+ * 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.example.springboot.numbers.mainrouter.config;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+import static org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.MODE_ALL_MATCH;
+import static org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.MODE_FIRST_MATCH;
+
+/**
+ * @param routingChannel The dynamic router channel.
+ * @param recipientMode The recipient mode -- first matching filter only, or all matching filters.
+ */
+@Validated
+@ConfigurationProperties(prefix = "main-router.dynamic-router-component")
+public record DynamicRouterComponentConfig(
+
+ @NotBlank String routingChannel,
+
+ @NotBlank @Pattern(regexp = "^" + MODE_FIRST_MATCH + "|" + MODE_ALL_MATCH + "$") String recipientMode) {
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/MainRouterConfig.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/MainRouterConfig.java
new file mode 100644
index 0000000..2cc4cfd
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/MainRouterConfig.java
@@ -0,0 +1,184 @@
+/*
+ * 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.example.springboot.numbers.mainrouter.config;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.CamelContext;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.dynamicrouter.control.DynamicRouterControlMessage;
+import org.apache.camel.example.springboot.numbers.mainrouter.model.StateMachineEvent;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.apache.camel.impl.engine.DefaultExecutorServiceManager;
+import org.apache.camel.support.DefaultThreadPoolFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+
+import static org.apache.camel.component.dynamicrouter.control.DynamicRouterControlConstants.COMPONENT_SCHEME_CONTROL;
+import static org.apache.camel.component.dynamicrouter.control.DynamicRouterControlConstants.CONTROL_ACTION_LIST;
+import static org.apache.camel.component.dynamicrouter.control.DynamicRouterControlConstants.CONTROL_ACTION_SUBSCRIBE;
+import static org.apache.camel.component.dynamicrouter.routing.DynamicRouterConstants.COMPONENT_SCHEME_ROUTING;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.ENDPOINT_DIRECT_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.ENDPOINT_DIRECT_LIST;
+import static org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events.PARTICIPANT_SUBSCRIBED;
+
+/**
+ * This configuration ingests the config properties in the application.yml file.
+ * Sets up the Camel route that feeds the Dynamic Router.
+ */
+@Configuration
+@EnableConfigurationProperties(DynamicRouterComponentConfig.class)
+public class MainRouterConfig {
+
+ /**
+ * Holds the config exchange.properties.
+ */
+ private final DynamicRouterComponentConfig dynamicRouterComponentConfig;
+
+ /**
+ * URI for sending command messages.
+ */
+ private final String commandUri;
+
+ /**
+ * URI for the dynamic router to send messages to.
+ */
+ private final String subscribeUri;
+
+ /**
+ * The Camel context.
+ */
+ private final CamelContext camelContext;
+
+ /**
+ * For publishing events.
+ */
+ final ApplicationEventPublisher eventPublisher;
+
+ /**
+ * Create this config with the config properties object.
+ *
+ * @param dynamicRouterComponentConfig the config properties object
+ * @param camelContext the Camel context
+ */
+ public MainRouterConfig(final DynamicRouterComponentConfig dynamicRouterComponentConfig,
+ @Value("${number-generator.command-uri}") String commandUri,
+ @Value("${number-generator.subscribe-uri}") String subscribeUri,
+ final CamelContext camelContext,
+ final ApplicationEventPublisher eventPublisher) {
+ this.dynamicRouterComponentConfig = dynamicRouterComponentConfig;
+ this.commandUri = commandUri;
+ this.subscribeUri = subscribeUri;
+ this.camelContext = camelContext;
+ this.eventPublisher = eventPublisher;
+ }
+
+ private static final Function<DynamicRouterComponentConfig, String> createCommandUri = cfg ->
+ "%s:%s?recipientMode=%s&executorService=%s".formatted(
+ COMPONENT_SCHEME_ROUTING,
+ cfg.routingChannel(),
+ cfg.recipientMode(),
+ "customPool");
+
+ @Bean
+ ObjectMapper objectMapper() {
+ return new ObjectMapper(new JsonFactory());
+ }
+
+ @Bean
+ public ExecutorService customPool() {
+ DefaultThreadPoolFactory threadPoolFactory = new DefaultThreadPoolFactory();
+ threadPoolFactory.setCamelContext(camelContext);
+ DefaultExecutorServiceManager executorServiceManager = new DefaultExecutorServiceManager(camelContext);
+ executorServiceManager.setThreadPoolFactory(threadPoolFactory);
+ return executorServiceManager.newCachedThreadPool(this, "DynamicRouterSpringBootThreadFactory");
+ }
+
+ /**
+ * Creates a simple route to allow a producer to send messages through
+ * the dynamic router on the routing channel.
+ */
+ @Bean
+ RouteBuilder numbersRouter() {
+ return new RouteBuilder(camelContext) {
+ @Override
+ public void configure() {
+ from(commandUri).to(createCommandUri.apply(dynamicRouterComponentConfig));
+ }
+ };
+ }
+
+ /**
+ * Creates a simple route to allow a producer to send messages through
+ * the dynamic router on the routing channel.
+ */
+ @Bean
+ RouteBuilder directCommandRouter() {
+ return new RouteBuilder(camelContext) {
+ @Override
+ public void configure() {
+ from(ENDPOINT_DIRECT_COMMAND).to(createCommandUri.apply(dynamicRouterComponentConfig));
+ }
+ };
+ }
+
+ /**
+ * Creates a simple route to allow a producer to send messages through
+ * the dynamic router on the routing channel.
+ */
+ @Bean
+ RouteBuilder listSubscribersRouter() {
+ return new RouteBuilder(camelContext) {
+ @Override
+ public void configure() {
+ from(ENDPOINT_DIRECT_LIST)
+ .toD(COMPONENT_SCHEME_CONTROL + ":" + CONTROL_ACTION_LIST +
+ "?subscribeChannel=${header.subscribeChannel}");
+ }
+ };
+ }
+
+ /**
+ * Creates a simple route to allow dynamic routing participants to
+ * subscribe or unsubscribe.
+ */
+ @Bean
+ RouteBuilder subscriptionRouter() {
+ return new RouteBuilder(camelContext) {
+ @Override
+ public void configure() {
+ from(subscribeUri)
+ .log(LoggingLevel.INFO, MainRouterConfig.class.getCanonicalName(), "Processing subscription: ${body}")
+ .unmarshal().json(DynamicRouterControlMessage.class)
+ .process(exchange -> {
+ Message<Events> message = MessageBuilder.withPayload(PARTICIPANT_SUBSCRIBED).build();
+ eventPublisher.publishEvent(new StateMachineEvent(this, message));
+ })
+ .to(COMPONENT_SCHEME_CONTROL + ":" + CONTROL_ACTION_SUBSCRIBE);
+ }
+ };
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineActionConfig.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineActionConfig.java
new file mode 100644
index 0000000..deb67ab
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineActionConfig.java
@@ -0,0 +1,73 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.config;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.service.NumberGeneratorService;
+import org.apache.camel.example.springboot.numbers.mainrouter.service.NumberStatisticsService;
+import org.apache.camel.example.springboot.numbers.mainrouter.service.WarmUpService;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.States;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.statemachine.action.Action;
+import org.springframework.statemachine.listener.StateMachineListener;
+import org.springframework.statemachine.listener.StateMachineListenerAdapter;
+import org.springframework.statemachine.state.State;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_EVENT_LIMIT;
+
+@Configuration
+public class StateMachineActionConfig {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateMachineActionConfig.class);
+
+ private final WarmUpService warmUpService;
+
+ private final NumberGeneratorService numberGeneratorService;
+
+ private final NumberStatisticsService numberStatisticsService;
+
+ public StateMachineActionConfig(WarmUpService warmUpService, NumberGeneratorService numberGeneratorService, NumberStatisticsService numberStatisticsService) {
+ this.warmUpService = warmUpService;
+ this.numberGeneratorService = numberGeneratorService;
+ this.numberStatisticsService = numberStatisticsService;
+ }
+
+ @Bean
+ public StateMachineListener<States, Events> listener() {
+ return new StateMachineListenerAdapter<>() {
+ @Override
+ public void stateChanged(State<States, Events> from, State<States, Events> to) {
+ LOG.info("State changed from '{}' to '{}'",
+ from == null ? "none" : from.getId(),
+ to.getId());
+ }
+ };
+ }
+
+ @Bean
+ public Action<States, Events> warmUpAction() {
+ return context -> {
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ executor.schedule(warmUpService::doWarmUp, 2, TimeUnit.SECONDS);
+ executor.shutdown();
+ };
+ }
+
+ @Bean
+ public Action<States, Events> generateNumbersAction() {
+ return context -> {
+ int limit = (int) context.getMessageHeader(HEADER_EVENT_LIMIT);
+ numberGeneratorService.generateNumbers(limit);
+ };
+ }
+
+ @Bean
+ public Action<States, Events> resetStatisticsAction() {
+ return context -> numberStatisticsService.resetStats();
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineConfig.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineConfig.java
new file mode 100644
index 0000000..9748391
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/config/StateMachineConfig.java
@@ -0,0 +1,89 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.config;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.States;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.statemachine.action.Action;
+import org.springframework.statemachine.config.EnableStateMachine;
+import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter;
+import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
+import org.springframework.statemachine.listener.StateMachineListener;
+
+import java.util.EnumSet;
+
+@Configuration
+@EnableStateMachine
+public class StateMachineConfig extends EnumStateMachineConfigurerAdapter<States, Events> {
+
+ private final StateMachineListener<States, Events> listener;
+
+ private final Action<States, Events> warmUpAction;
+
+ private final Action<States, Events> generateNumbersAction;
+
+ private final Action<States, Events> resetStatisticsAction;
+
+ private boolean initialized = false;
+
+ public StateMachineConfig(StateMachineListener<States, Events> listener,
+ @Qualifier("warmUpAction") Action<States, Events> warmUpAction,
+ @Qualifier("generateNumbersAction") Action<States, Events> generateNumbersAction,
+ @Qualifier("resetStatisticsAction") Action<States, Events> resetStatisticsAction) {
+ this.listener = listener;
+ this.warmUpAction = warmUpAction;
+ this.generateNumbersAction = generateNumbersAction;
+ this.resetStatisticsAction = resetStatisticsAction;
+ }
+
+ @Override
+ public void configure(StateMachineConfigurationConfigurer<States, Events> config) throws Exception {
+ config
+ .withConfiguration()
+ .autoStartup(true)
+ .listener(listener);
+ }
+
+ @Override
+ public void configure(StateMachineStateConfigurer<States, Events> states) throws Exception {
+ states
+ .withStates()
+ .initial(States.STARTING)
+ .states(EnumSet.allOf(States.class));
+ }
+
+ @Override
+ public void configure(StateMachineTransitionConfigurer<States, Events> transitions) throws Exception {
+ // @formatter:off
+ transitions
+ .withExternal()
+ .source(States.STARTING)
+ .target(States.INITIALIZING)
+ .event(Events.PARTICIPANT_SUBSCRIBED)
+ .guard(context -> !initialized)
+ .action(context -> initialized = true)
+ .action(resetStatisticsAction)
+ .action(warmUpAction)
+ .and()
+ .withExternal()
+ .source(States.INITIALIZING)
+ .target(States.READY)
+ .event(Events.INITIALIZATION_COMPLETE)
+ .action(resetStatisticsAction)
+ .and()
+ .withExternal()
+ .source(States.READY)
+ .target(States.GENERATING_NUMBERS)
+ .event(Events.GENERATE_NUMBERS_STARTED)
+ .action(resetStatisticsAction)
+ .action(generateNumbersAction)
+ .and()
+ .withExternal()
+ .source(States.GENERATING_NUMBERS)
+ .target(States.READY)
+ .event(Events.GENERATE_NUMBERS_COMPLETE);
+ // @formatter:on
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/model/StateMachineEvent.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/model/StateMachineEvent.java
new file mode 100644
index 0000000..2075f74
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/model/StateMachineEvent.java
@@ -0,0 +1,19 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.model;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.messaging.Message;
+
+public class StateMachineEvent extends ApplicationEvent {
+
+ private final transient Message<Events> message;
+
+ public StateMachineEvent(Object source, Message<Events> message) {
+ super(source);
+ this.message = message;
+ }
+
+ public Message<Events> getMessage() {
+ return message;
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/routing/NumberStatisticsRoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/routing/NumberStatisticsRoutingParticipant.java
new file mode 100644
index 0000000..eb489b3
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/routing/NumberStatisticsRoutingParticipant.java
@@ -0,0 +1,75 @@
+/*
+ * 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.example.springboot.numbers.mainrouter.routing;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consume;
+import org.apache.camel.Header;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.dynamicrouter.control.DynamicRouterControlMessage;
+import org.apache.camel.example.springboot.numbers.common.service.RoutingParticipant;
+import org.apache.camel.example.springboot.numbers.mainrouter.service.NumberStatisticsService;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import static org.apache.camel.component.dynamicrouter.control.DynamicRouterControlConstants.COMPONENT_SCHEME_CONTROL;
+import static org.apache.camel.component.dynamicrouter.control.DynamicRouterControlConstants.CONTROL_ACTION_SUBSCRIBE;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_NUMBER;
+
+@Service
+public class NumberStatisticsRoutingParticipant extends RoutingParticipant {
+
+ private final NumberStatisticsService numberStatisticsService;
+
+ public NumberStatisticsRoutingParticipant(final NumberStatisticsService numberStatisticsService,
+ @Value("${number-generator.subscribe-uri}") String subscribeUri,
+ @Value("${number-generator.routing-channel}") String routingChannel,
+ @Value("${number-generator.predicate}") String predicate,
+ @Value("${number-generator.expression-language}") String expressionLanguage,
+ @Value("${number-generator.subscription-priority}") int subscriptionPriority,
+ @Value("${number-generator.consume-uri}") String consumeUri,
+ @Value("${number-generator.command-uri}") String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ super("processNumberStats", subscribeUri, routingChannel, subscriptionPriority,
+ predicate, expressionLanguage, consumeUri, commandUri, producerTemplate, camelContext);
+ this.numberStatisticsService = numberStatisticsService;
+ }
+
+ /**
+ * Send the subscribe message after this service instance is created.
+ */
+ @Override
+ protected void subscribe() {
+ DynamicRouterControlMessage message = createSubscribeMessage();
+ producerTemplate.sendBody(COMPONENT_SCHEME_CONTROL + ":" + CONTROL_ACTION_SUBSCRIBE, message);
+ }
+
+ /**
+ * This method consumes messages that have matched the participant's rules
+ * and have been routed to the participant. It adds the results to the
+ * results service.
+ *
+ * @param body the serialized command message
+ */
+ @Override
+ @Consume(property = "consumeUri")
+ public void consumeMessage(final String body, @Header(value = HEADER_NUMBER) String number) {
+ this.numberStatisticsService.updateStats(body);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberGeneratorService.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberGeneratorService.java
new file mode 100644
index 0000000..2ea265d
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberGeneratorService.java
@@ -0,0 +1,70 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.service;
+
+import jakarta.validation.constraints.NotNull;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.mainrouter.model.StateMachineEvent;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+import java.util.Map;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.COMMAND_PROCESS_NUMBER;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.ENDPOINT_DIRECT_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_NUMBER;
+import static org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events.GENERATE_NUMBERS_COMPLETE;
+
+@Service
+public class NumberGeneratorService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NumberGeneratorService.class);
+
+ private final ProducerTemplate producerTemplate;
+
+ private final ApplicationEventPublisher eventPublisher;
+
+ public NumberGeneratorService(@NotNull final ProducerTemplate producerTemplate,
+ ApplicationEventPublisher eventPublisher) {
+ this.producerTemplate = producerTemplate;
+ this.eventPublisher = eventPublisher;
+ producerTemplate.start();
+ }
+
+ /**
+ * When a command has been received to generate numbers, this will continuously generate
+ * numbers and send them in a command to have recipients process the numbers. It will
+ * only stop when a limit (if any) is reached, or if a subsequent command instructs
+ * number message generation to stop
+ *
+ * @param limit the count of numbers to produce (zero means Integer.MAX_VALUE)
+ */
+ public void generateNumbers(int limit) {
+ String msg;
+ try {
+ LOG.info("Generating numbers from 1 to {}", limit);
+ Flux.range(1, limit)
+ .flatMap(n -> Mono.just(n)
+ .map(Object::toString)
+ .subscribeOn(Schedulers.boundedElastic())
+ .doOnNext(strN -> producerTemplate.sendBodyAndHeaders(ENDPOINT_DIRECT_COMMAND, strN,
+ Map.of(HEADER_COMMAND, COMMAND_PROCESS_NUMBER, HEADER_NUMBER, strN))))
+ .doFinally(x -> {
+ Message<MainRouterUtil.Events> message =
+ MessageBuilder.withPayload(GENERATE_NUMBERS_COMPLETE).build();
+ eventPublisher.publishEvent(new StateMachineEvent(this, message));
+ })
+ .subscribe();
+ } catch (Exception e) {
+ msg = String.format("Exception when trying to send number messages: %s", e.getMessage());
+ LOG.warn(msg, e);
+ }
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberStatisticsService.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberStatisticsService.java
new file mode 100644
index 0000000..03c6aa0
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/NumberStatisticsService.java
@@ -0,0 +1,66 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.common.model.CommandMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.COMMAND_RESET_STATS;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.ENDPOINT_DIRECT_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_COMMAND;
+
+@Service
+public class NumberStatisticsService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NumberStatisticsService.class);
+
+ private final Map<String, Long> countsMap;
+
+ private final AtomicLong currentMs = new AtomicLong(0L);
+
+ private long start = 0L;
+
+ private final ProducerTemplate producerTemplate;
+
+ private final ObjectMapper objectMapper;
+
+ public NumberStatisticsService(final ProducerTemplate producerTemplate,
+ final ObjectMapper objectMapper) {
+ this.producerTemplate = producerTemplate;
+ this.countsMap = new ConcurrentSkipListMap<>();
+ this.objectMapper = objectMapper;
+ }
+
+ public String resetStats() {
+ start = System.currentTimeMillis();
+ countsMap.clear();
+ currentMs.set(0L);
+ producerTemplate.sendBodyAndHeader(ENDPOINT_DIRECT_COMMAND, COMMAND_RESET_STATS, HEADER_COMMAND, COMMAND_RESET_STATS);
+ return "OK - Statistics have been reset.";
+ }
+
+ public void updateStats(final String body) {
+ try {
+ CommandMessage message = objectMapper.readValue(body, CommandMessage.class);
+ long now = System.currentTimeMillis();
+ long newTime = now - start;
+ long elapsed = currentMs.updateAndGet(cv -> Math.max(cv, newTime));
+ countsMap.put("elapsed seconds", elapsed / 1000);
+ message.params().forEach((key, val) -> countsMap.merge(key, Long.parseLong(val), Math::max));
+ } catch (JsonProcessingException e) {
+ LOG.warn("Error when trying to update number statistics", e);
+ }
+ }
+
+ public Map<String, Long> getCounts() {
+ LOG.info("Getting number statistics");
+ return Map.copyOf(countsMap);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/StateMachineEventTriggerService.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/StateMachineEventTriggerService.java
new file mode 100644
index 0000000..f596da8
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/StateMachineEventTriggerService.java
@@ -0,0 +1,27 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.service;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.model.StateMachineEvent;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.event.EventListener;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.stereotype.Service;
+
+@Service
+public class StateMachineEventTriggerService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateMachineEventTriggerService.class);
+
+ private final StateMachine<MainRouterUtil.States, MainRouterUtil.Events> stateMachine;
+
+ public StateMachineEventTriggerService(StateMachine<MainRouterUtil.States, MainRouterUtil.Events> stateMachine) {
+ this.stateMachine = stateMachine;
+ }
+
+ @EventListener
+ public void onEvent(StateMachineEvent event) {
+ LOG.info("Received state machine event: {}", event.getMessage().getPayload().name());
+ MainRouterUtil.sendEventAndMapResponse.apply(stateMachine, event.getMessage()).subscribe();
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/SubscribersService.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/SubscribersService.java
new file mode 100644
index 0000000..1d7d9f8
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/SubscribersService.java
@@ -0,0 +1,25 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.service;
+
+import org.apache.camel.ProducerTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+@Service
+public class SubscribersService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SubscribersService.class);
+
+ private final ProducerTemplate producerTemplate;
+
+ public SubscribersService(final ProducerTemplate producerTemplate) {
+ this.producerTemplate = producerTemplate;
+ producerTemplate.start();
+ }
+
+ public String listSubscriptions(final String channel) {
+ LOG.info("Getting subscriptions list for channel '{}'", channel);
+ return producerTemplate.requestBodyAndHeader(
+ "direct:list", "", "subscribeChannel", channel, String.class);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/WarmUpService.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/WarmUpService.java
new file mode 100644
index 0000000..99bb717
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/service/WarmUpService.java
@@ -0,0 +1,63 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.service;
+
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.mainrouter.model.StateMachineEvent;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+import java.util.Map;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.COMMAND_PROCESS_NUMBER;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.ENDPOINT_DIRECT_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_NUMBER;
+import static org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events.INITIALIZATION_COMPLETE;
+
+/**
+ * This service sends a small number of messages through the system shortly after startup to
+ * warm up the JVM so that we can see optimal results the first time we run a batch of
+ * messages through the system.
+ */
+@Service
+public class WarmUpService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WarmUpService.class);
+
+ private final ProducerTemplate producerTemplate;
+
+ private final ApplicationEventPublisher eventPublisher;
+
+ public WarmUpService(ProducerTemplate producerTemplate,
+ ApplicationEventPublisher eventPublisher) {
+ this.producerTemplate = producerTemplate;
+ this.eventPublisher = eventPublisher;
+ }
+
+ /**
+ * Sends messages for the warm-up.
+ */
+ public void doWarmUp() {
+ LOG.info("Running warm-up...");
+ Flux.range(1, 1000000)
+ .flatMap(n -> Mono.just(n)
+ .subscribeOn(Schedulers.boundedElastic())
+ .map(String::valueOf)
+ .doOnNext(ns -> producerTemplate.sendBodyAndHeaders(ENDPOINT_DIRECT_COMMAND, 1,
+ Map.of(HEADER_COMMAND, COMMAND_PROCESS_NUMBER,
+ HEADER_NUMBER, ns))))
+ .doFinally(x -> {
+ LOG.info("Warm-up finished");
+ Message<Events> message = MessageBuilder.withPayload(INITIALIZATION_COMPLETE).build();
+ eventPublisher.publishEvent(new StateMachineEvent(this, message));
+ })
+ .subscribe();
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/util/MainRouterUtil.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/util/MainRouterUtil.java
new file mode 100644
index 0000000..5ac2b61
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/util/MainRouterUtil.java
@@ -0,0 +1,50 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.util;
+
+import org.springframework.messaging.Message;
+import org.springframework.statemachine.StateMachine;
+import reactor.core.publisher.Mono;
+
+import java.util.function.BiFunction;
+
+public class MainRouterUtil {
+
+ public static final BiFunction<StateMachine<States, Events>, Message<Events>, Mono<String>> sendEventAndMapResponse = (sm, m) ->
+ sm.sendEvent(Mono.just(m))
+ .next()
+ .map(er ->
+ switch (er.getResultType()) {
+ case ACCEPTED -> "Event accepted.";
+ case DENIED -> "Event denied.";
+ case DEFERRED -> "Event deferred.";
+ });
+
+ public enum States {
+ STARTING("Starting"),
+ INITIALIZING("Initializing"),
+ READY("Ready"),
+ GENERATING_NUMBERS("Generating Numbers");
+
+ private final String description;
+
+ States(String description) {
+ this.description = description;
+ }
+
+ public String getDescription() {
+ return this.description;
+ }
+ }
+
+ public enum Events {
+ PARTICIPANT_SUBSCRIBED("Routing Participant Subscribed"),
+ INITIALIZATION_COMPLETE("Initialization Complete"),
+ GENERATE_NUMBERS_STARTED("Generate Numbers Started"),
+ GENERATE_NUMBERS_COMPLETE("Generate Numbers Complete");
+
+ private final String description;
+
+ Events(String description) {
+ this.description = description;
+ }
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberGeneratorController.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberGeneratorController.java
new file mode 100644
index 0000000..0e57754
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberGeneratorController.java
@@ -0,0 +1,33 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.web;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events;
+import org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.States;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_EVENT_LIMIT;
+import static org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.Events.GENERATE_NUMBERS_STARTED;
+import static org.apache.camel.example.springboot.numbers.mainrouter.util.MainRouterUtil.sendEventAndMapResponse;
+
+@RestController
+public class NumberGeneratorController {
+
+ final StateMachine<States, Events> stateMachine;
+
+ public NumberGeneratorController(StateMachine<States, Events> stateMachine) {
+ this.stateMachine = stateMachine;
+ }
+
+ @PutMapping(path = "/generate")
+ public Mono<String> generate(@RequestParam("limit") int limit) {
+ Message<Events> message = MessageBuilder.withPayload(GENERATE_NUMBERS_STARTED)
+ .setHeader(HEADER_EVENT_LIMIT, limit)
+ .build();
+ return sendEventAndMapResponse.apply(stateMachine, message);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberStatisticsController.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberStatisticsController.java
new file mode 100644
index 0000000..b50559a
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/NumberStatisticsController.java
@@ -0,0 +1,40 @@
+/*
+ * 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.example.springboot.numbers.mainrouter.web;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.service.NumberStatisticsService;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+
+@RestController
+public class NumberStatisticsController {
+
+ private final NumberStatisticsService numberStatisticsService;
+
+ public NumberStatisticsController(NumberStatisticsService numberStatisticsService) {
+ this.numberStatisticsService = numberStatisticsService;
+ }
+
+ @GetMapping(path = "/counts")
+ public Mono<Map<String, Long>> getCounts() {
+ return Mono.just(numberStatisticsService.getCounts());
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/SubscribersController.java b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/SubscribersController.java
new file mode 100644
index 0000000..8e5ef6f
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/java/org/apache/camel/example/springboot/numbers/mainrouter/web/SubscribersController.java
@@ -0,0 +1,22 @@
+package org.apache.camel.example.springboot.numbers.mainrouter.web;
+
+import org.apache.camel.example.springboot.numbers.mainrouter.service.SubscribersService;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+@RestController
+public class SubscribersController {
+
+ private final SubscribersService subscribersService;
+
+ public SubscribersController(SubscribersService subscribersService) {
+ this.subscribersService = subscribersService;
+ }
+
+ @GetMapping(path = "/list/{channel}")
+ public Mono<String> listSubscriptions(@PathVariable("channel") String channel) {
+ return Mono.just(subscribersService.listSubscriptions(channel));
+ }
+}
diff --git a/dynamic-router-eip/src/test/resources/application-test-less.yml b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/resources/application.yaml
similarity index 63%
copy from dynamic-router-eip/src/test/resources/application-test-less.yml
copy to dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/resources/application.yaml
index 6d43362..fea4351 100644
--- a/dynamic-router-eip/src/test/resources/application-test-less.yml
+++ b/dynamic-router-eip/dynamic-router-eip-stack/main-router/src/main/resources/application.yaml
@@ -1,45 +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.
-#
-camel:
- springboot:
- name: CamelSpringBootDynamicRouterExample
- jmx-enabled: false
- shutdown-timeout: 30
- spring-boot:
- example:
- dynamic-router-eip:
- routing-channel: numbers
- subscribe-uri: dynamic-router:control
- receiver-scheme: direct
- start-uri: direct:start
- recipient-mode: firstMatch
- send-message-count: 10000
- expected-first-match-message-count: 1000
- expected-all-match-message-count: 500000
- component:
- dynamic-router:
- lazy-start-producer: true
- bridge-error-handler: true
- cloud:
- enabled: false
-logging:
- level:
- root: WARN
- org.apache.camel.example.springboot: INFO
-spring:
- main:
- banner-mode: off
+#
+# 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.
+#
+app:
+ name: "main-router"
+ port: 8082
+ id: "main_router"
+ predicate: "{headers.command == 'stats'}"
+ priority: 10
+camel:
+ springboot:
+ name: CamelSpringBootDynamicRouterExampleMainRouter
+ jmx-enabled: false
+ shutdown-timeout: 30
+ endpoint-bridge-error-handler: true
+ component:
+ dynamic-router:
+ enabled: true
+ lazy-start-producer: true
+ spring-event:
+ bridge-error-handler: true
+ kafka:
+ bridge-error-handler: true
+main-router:
+ dynamic-router-component:
+ routing-channel: "numbers"
+ recipient-mode: "allMatch"
+logging:
+ level:
+ root: WARN
+ org.apache.kafka: ERROR
+ org.apache.camel: INFO
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/pom.xml
new file mode 100644
index 0000000..0ea4629
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/pom.xml
@@ -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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>numbers-common</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule :: Common Library</name>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <version>${spring-boot-version}</version>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/CommonConfig.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/CommonConfig.java
new file mode 100644
index 0000000..c7a9a8b
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/CommonConfig.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.example.springboot.numbers.common.config;
+
+import org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer;
+import org.apache.camel.example.springboot.numbers.common.service.StringValueHeaderDeserializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@EnableScheduling
+@Configuration
+@PropertySource(value = "classpath:application.yaml", factory = YamlPropertySourceFactory.class, ignoreResourceNotFound = true)
+@PropertySource(value = "classpath:common.yaml", factory = YamlPropertySourceFactory.class, ignoreResourceNotFound = true)
+public class CommonConfig {
+
+ @Bean
+ KafkaHeaderDeserializer stringValueHeaderDeserializer() {
+ return new StringValueHeaderDeserializer();
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/YamlPropertySourceFactory.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/YamlPropertySourceFactory.java
new file mode 100644
index 0000000..d0f260e
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/config/YamlPropertySourceFactory.java
@@ -0,0 +1,28 @@
+package org.apache.camel.example.springboot.numbers.common.config;
+
+import org.springframework.core.io.Resource;
+import org.springframework.lang.NonNull;
+import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
+import org.springframework.core.env.PropertiesPropertySource;
+import org.springframework.core.env.PropertySource;
+import org.springframework.core.io.support.EncodedResource;
+import org.springframework.core.io.support.PropertySourceFactory;
+
+import java.util.Optional;
+import java.util.Properties;
+
+public class YamlPropertySourceFactory implements PropertySourceFactory {
+
+ @Override
+ @NonNull
+ public PropertySource<?> createPropertySource(String name, EncodedResource encodedResource) {
+ YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
+ Resource resource = encodedResource.getResource();
+ String resourceFileName = Optional.ofNullable(resource.getFilename())
+ .orElseThrow(() -> new IllegalArgumentException("Resource file name must not be null"));
+ factory.setResources(resource);
+ Properties properties = Optional.ofNullable(factory.getObject())
+ .orElseThrow(() -> new IllegalArgumentException("Properties factory object must not be null"));
+ return new PropertiesPropertySource(resourceFileName, properties);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/model/CommandMessage.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/model/CommandMessage.java
new file mode 100644
index 0000000..f6cfb31
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/model/CommandMessage.java
@@ -0,0 +1,34 @@
+package org.apache.camel.example.springboot.numbers.common.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+public record CommandMessage(String command, Map<String, String> params) {
+
+ @JsonIgnore
+ private static final AtomicInteger errorCount = new AtomicInteger(0);
+
+ @JsonIgnore
+ private static final ObjectMapper objectMapper = new ObjectMapper(new JsonFactory());
+
+ @Override
+ public String toString() {
+ String result;
+ try {
+ result = objectMapper.writeValueAsString(this);
+ } catch (Exception ex) {
+ result = Map.of("errorCount", String.valueOf(errorCount.incrementAndGet()),
+ "errorMessage", ex.getMessage()).entrySet().stream()
+ .map(e -> String.format("""
+ "%s":"%s"
+ """, e.getKey(), e.getValue()))
+ .collect(Collectors.joining(",", "{", "}"));
+ }
+ return result;
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/ProcessNumbersRoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/ProcessNumbersRoutingParticipant.java
new file mode 100644
index 0000000..0005442
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/ProcessNumbersRoutingParticipant.java
@@ -0,0 +1,94 @@
+/*
+ * 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.example.springboot.numbers.common.service;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consume;
+import org.apache.camel.Header;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.common.model.CommandMessage;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.COMMAND_RESET_STATS;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.COMMAND_STATS;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_COMMAND;
+import static org.apache.camel.example.springboot.numbers.common.util.NumbersCommonUtil.HEADER_NUMBER;
+
+public abstract class ProcessNumbersRoutingParticipant extends RoutingParticipant {
+
+ protected final String numberName;
+
+ /**
+ * Counter for the number of messages that this participant has processed.
+ */
+ protected final AtomicInteger processedCount = new AtomicInteger(0);
+
+ /**
+ * The count that was last reported via "stats" command message.
+ */
+ protected final AtomicInteger reportedCount = new AtomicInteger(0);
+
+ protected ProcessNumbersRoutingParticipant( // NOSONAR
+ String numberName,
+ String subscriberId,
+ String subscribeUri,
+ String routingChannel,
+ int subscriptionPriority,
+ String predicate,
+ String expressionLanguage,
+ String consumeUri,
+ String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ super(subscriberId, subscribeUri, routingChannel, subscriptionPriority, predicate, expressionLanguage,
+ consumeUri, commandUri, producerTemplate, camelContext);
+ this.numberName = numberName;
+ producerTemplate.start();
+ }
+
+ /**
+ * This method consumes messages that have matched the participant's rules
+ * and have been routed to the participant.
+ *
+ * @param body the serialized command message
+ */
+ @Override
+ @Consume(property = "consumeUri")
+ public void consumeMessage(final String body, @Header(value = HEADER_NUMBER) String number) {
+ if (COMMAND_RESET_STATS.equals(body)) {
+ processedCount.set(0);
+ reportedCount.set(0);
+ } else {
+ processedCount.incrementAndGet();
+ }
+ }
+
+ @Scheduled(fixedRate = 2, timeUnit = TimeUnit.SECONDS)
+ public void sendStats() {
+ int pCount = processedCount.get();
+ if (pCount > reportedCount.get()) {
+ CommandMessage command = new CommandMessage(COMMAND_STATS, Map.of(numberName, String.valueOf(pCount)));
+ producerTemplate.sendBodyAndHeader(commandUri, command.toString(), HEADER_COMMAND, COMMAND_STATS);
+ reportedCount.set(pCount);
+ }
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/RoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/RoutingParticipant.java
new file mode 100644
index 0000000..1226311
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/RoutingParticipant.java
@@ -0,0 +1,166 @@
+/*
+ * 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.example.springboot.numbers.common.service;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Header;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.dynamicrouter.control.DynamicRouterControlMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+
+public abstract class RoutingParticipant {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(RoutingParticipant.class);
+
+ private final ObjectMapper objectMapper = new ObjectMapper(new JsonFactory());
+
+ protected final String subscriberId;
+
+ /**
+ * The dynamic router control channel URI where subscribe messages will
+ * be sent.
+ */
+ protected final String subscribeUri;
+
+ /**
+ * The channel of the dynamic router to send messages.
+ */
+ protected final String routingChannel;
+
+ /**
+ * The priority of the processor when evaluated by the dynamic router. Lower
+ * number means higher priority.
+ */
+ protected final int priority;
+
+ /**
+ * The predicate by which exchanges are evaluated for suitability for
+ * a routing participant.
+ */
+ protected final String predicate;
+
+ /**
+ * The language of the predicate string.
+ */
+ protected final String expressionLanguage;
+
+ /**
+ * The URI that a participant implementation will listen on for messages
+ * that match its rules.
+ */
+ protected final String consumeUri;
+
+ /**
+ * URI to send a command to (for dynamic routing).
+ */
+ protected final String commandUri;
+
+ /**
+ * The {@link ProducerTemplate} to send subscriber messages to the dynamic
+ * router control channel.
+ */
+ protected final ProducerTemplate producerTemplate;
+
+ protected final CamelContext camelContext;
+
+ protected RoutingParticipant( // NOSONAR
+ String subscriberId,
+ String subscribeUri,
+ String routingChannel,
+ int subscriptionPriority,
+ String predicate,
+ String expressionLanguage,
+ String consumeUri,
+ String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ this.subscriberId = subscriberId;
+ this.subscribeUri = subscribeUri;
+ this.routingChannel = routingChannel;
+ this.priority = subscriptionPriority;
+ this.predicate = predicate;
+ this.expressionLanguage = expressionLanguage;
+ this.consumeUri = consumeUri;
+ this.commandUri = commandUri;
+ this.producerTemplate = producerTemplate;
+ this.camelContext = camelContext;
+ producerTemplate.start();
+ }
+
+ /**
+ * Send the subscribe message after this service instance is created.
+ */
+ protected void subscribe() {
+ try {
+ DynamicRouterControlMessage message = createSubscribeMessage();
+ String messageJson = objectMapper.writeValueAsString(message);
+ LOG.info("Sending subscribe message: {} to {}", messageJson, subscribeUri);
+ producerTemplate.sendBody(subscribeUri, messageJson);
+ } catch (Exception e) {
+ throw new IllegalStateException("Could not serialize a message", e);
+ }
+ }
+
+ /**
+ * After the application is started and ready, subscribe for messages.
+ */
+ @EventListener(ApplicationReadyEvent.class)
+ protected void start() {
+ subscribe();
+ }
+
+ /**
+ * This method consumes messages that have matched the participant's rules
+ * and have been routed to the participant. It adds the results to the
+ * results service.
+ *
+ * @param body the serialized command message
+ */
+ public abstract void consumeMessage(final String body, @Header(value = "number") String number);
+
+ /**
+ * Create a {@link DynamicRouterControlMessage} based on parameters from the
+ * implementing class.
+ *
+ * @return the {@link DynamicRouterControlMessage}
+ */
+ protected DynamicRouterControlMessage createSubscribeMessage() {
+ return DynamicRouterControlMessage.Builder.newBuilder()
+ .subscribeChannel(routingChannel)
+ .subscriptionId(subscriberId)
+ .destinationUri(consumeUri)
+ .priority(priority)
+ .predicate("#" + predicate)
+ .expressionLanguage(expressionLanguage)
+ .build();
+ }
+
+ /**
+ * Gets the consumer URI.
+ *
+ * @return the consumer URI
+ */
+ public String getConsumeUri() {
+ return this.consumeUri;
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/StringValueHeaderDeserializer.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/StringValueHeaderDeserializer.java
new file mode 100644
index 0000000..2d0f825
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/service/StringValueHeaderDeserializer.java
@@ -0,0 +1,30 @@
+/*
+ * 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.example.springboot.numbers.common.service;
+
+import org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer;
+
+import java.nio.charset.StandardCharsets;
+
+public class StringValueHeaderDeserializer implements KafkaHeaderDeserializer {
+
+ @Override
+ public Object deserialize(String key, byte[] value) {
+ return new String(value, StandardCharsets.UTF_8);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/util/NumbersCommonUtil.java b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/util/NumbersCommonUtil.java
new file mode 100644
index 0000000..0dbcfbd
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/java/org/apache/camel/example/springboot/numbers/common/util/NumbersCommonUtil.java
@@ -0,0 +1,25 @@
+package org.apache.camel.example.springboot.numbers.common.util;
+
+/**
+ * Common constants and methods.
+ */
+public abstract class NumbersCommonUtil {
+
+ private NumbersCommonUtil() {}
+
+ public static final String ENDPOINT_DIRECT_COMMAND = "direct:command";
+
+ public static final String ENDPOINT_DIRECT_LIST = "direct:list";
+
+ public static final String COMMAND_PROCESS_NUMBER = "processNumber";
+
+ public static final String COMMAND_RESET_STATS = "resetStats";
+
+ public static final String COMMAND_STATS = "stats";
+
+ public static final String HEADER_COMMAND = "command";
+
+ public static final String HEADER_EVENT_LIMIT = "limit";
+
+ public static final String HEADER_NUMBER = "number";
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/resources/common.yaml b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/resources/common.yaml
new file mode 100644
index 0000000..e09004a
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/numbers-common/src/main/resources/common.yaml
@@ -0,0 +1,54 @@
+#
+# 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.
+#
+spring:
+ application:
+ name: "${app.name}"
+ webflux:
+ base-path: "/${app.name}"
+server:
+ port: ${app.port}
+management:
+ endpoints:
+ enabled-by-default: false
+ web:
+ exposure:
+ include: "health"
+ endpoint:
+ health:
+ enabled: true
+camel:
+ component:
+ kafka:
+ brokers: "broker:9092"
+ partitioner: "org.apache.kafka.clients.producer.RoundRobinPartitioner"
+ groupId: "numbers"
+ bridge-error-handler: true
+ jmx:
+ disabled: true
+number-generator:
+ routing-channel: "numbers"
+ subscribe-uri: "kafka://control?groupInstanceId=${app.id}_subscribe&headerDeserializer=#stringValueHeaderDeserializer&partitioner=org.apache.kafka.clients.producer.RoundRobinPartitioner"
+ command-uri: "kafka://numbers_command?groupInstanceId=${app.id}_command&headerDeserializer=#stringValueHeaderDeserializer&partitioner=org.apache.kafka.clients.producer.RoundRobinPartitioner"
+ consume-uri: "kafka://${app.id}?groupInstanceId=${app.id}_consumer&headerDeserializer=#stringValueHeaderDeserializer"
+ predicate: "${app.predicate}"
+ expression-language: "spel"
+ subscription-priority: ${app.priority}
+logging:
+ level:
+ root: WARN
+ org.apache.kafka: ERROR
+ org.apache.camel: INFO
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/pom.xml
new file mode 100644
index 0000000..dd14a31
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/pom.xml
@@ -0,0 +1,41 @@
+<!--
+ ~ 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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>odd-numbers-service</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule :: Odd Numbers Service</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>numbers-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/Application.java b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/Application.java
new file mode 100644
index 0000000..40c8120
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/Application.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.example.springboot.numbers.odd;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "org.apache.camel.example.springboot.numbers")
+public class Application {
+
+ /**
+ * Main method to start the application. Please make us proud.
+ */
+ public static void main(String[] args) {
+ SpringApplication.run(Application.class, args);
+ }
+}
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/service/ProcessOddNumbersRoutingParticipant.java b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/service/ProcessOddNumbersRoutingParticipant.java
new file mode 100644
index 0000000..415406c
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/java/org/apache/camel/example/springboot/numbers/odd/service/ProcessOddNumbersRoutingParticipant.java
@@ -0,0 +1,42 @@
+/*
+ * 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.example.springboot.numbers.odd.service;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.example.springboot.numbers.common.service.ProcessNumbersRoutingParticipant;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ProcessOddNumbersRoutingParticipant extends ProcessNumbersRoutingParticipant {
+
+ public ProcessOddNumbersRoutingParticipant(
+ @Value("${number-generator.subscribe-uri}") String subscribeUri,
+ @Value("${number-generator.routing-channel}") String routingChannel,
+ @Value("${number-generator.predicate}") String predicate,
+ @Value("${number-generator.expression-language}") String expressionLanguage,
+ @Value("${number-generator.subscription-priority}") int subscriptionPriority,
+ @Value("${number-generator.consume-uri}") String consumeUri,
+ @Value("${number-generator.command-uri}") String commandUri,
+ ProducerTemplate producerTemplate,
+ CamelContext camelContext) {
+ super("odd", "processOddNumbers", subscribeUri, routingChannel, subscriptionPriority,
+ predicate, expressionLanguage, consumeUri, commandUri, producerTemplate, camelContext);
+ }
+}
diff --git a/dynamic-router-eip/src/test/resources/application-test-less.yml b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/resources/application.yaml
similarity index 50%
rename from dynamic-router-eip/src/test/resources/application-test-less.yml
rename to dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/resources/application.yaml
index 6d43362..84a25e9 100644
--- a/dynamic-router-eip/src/test/resources/application-test-less.yml
+++ b/dynamic-router-eip/dynamic-router-eip-stack/odd-numbers-service/src/main/resources/application.yaml
@@ -1,45 +1,22 @@
-#
-# 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.
-#
-camel:
- springboot:
- name: CamelSpringBootDynamicRouterExample
- jmx-enabled: false
- shutdown-timeout: 30
- spring-boot:
- example:
- dynamic-router-eip:
- routing-channel: numbers
- subscribe-uri: dynamic-router:control
- receiver-scheme: direct
- start-uri: direct:start
- recipient-mode: firstMatch
- send-message-count: 10000
- expected-first-match-message-count: 1000
- expected-all-match-message-count: 500000
- component:
- dynamic-router:
- lazy-start-producer: true
- bridge-error-handler: true
- cloud:
- enabled: false
-logging:
- level:
- root: WARN
- org.apache.camel.example.springboot: INFO
-spring:
- main:
- banner-mode: off
+#
+# 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.
+#
+app:
+ name: "odd-numbers"
+ port: 8901
+ id: "numbers_odd"
+ predicate: "{(headers.command == 'processNumber' and headers.number matches '\\d*[13579]') or headers.command == 'resetStats'}"
+ priority: 10
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/pom.xml b/dynamic-router-eip/dynamic-router-eip-stack/pom.xml
new file mode 100644
index 0000000..652112a
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/pom.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel.springboot.example</groupId>
+ <artifactId>camel-example-spring-boot-dynamic-router-eip</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-example-spring-boot-dynamic-router-eip-stack</artifactId>
+ <packaging>pom</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples :: Multimodule</name>
+ <description>Dynamic Router EIP component example in Spring Boot with multiple modules</description>
+
+ <modules>
+ <module>numbers-common</module>
+ <module>main-router</module>
+ <module>all-numbers-service</module>
+ <module>even-numbers-service</module>
+ <module>odd-numbers-service</module>
+ </modules>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-actuator</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-kafka-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.statemachine</groupId>
+ <artifactId>spring-statemachine-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-jackson-starter</artifactId>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/dynamic-router-eip/dynamic-router-eip-stack/project-resources/docker/docker-compose.yaml b/dynamic-router-eip/dynamic-router-eip-stack/project-resources/docker/docker-compose.yaml
new file mode 100644
index 0000000..66cc8c9
--- /dev/null
+++ b/dynamic-router-eip/dynamic-router-eip-stack/project-resources/docker/docker-compose.yaml
@@ -0,0 +1,141 @@
+#
+# 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.
+#
+version: "3.7"
+
+services:
+ redpanda_broker:
+ image: docker.io/vectorized/redpanda:v23.2.19
+ container_name: broker
+ networks:
+ - test-dynamic-router
+ volumes:
+ - broker:/var/lib/redpanda/data:rw
+ user: "101:101"
+ cap_add:
+ - SYS_NICE
+ privileged: true
+ command:
+ - redpanda start
+ - --overprovisioned
+ - --smp 1
+ - --memory 1G
+ - --reserve-memory 0M
+ - --kafka-addr internal://broker:9092
+ - --advertise-kafka-addr internal://broker:9092
+ - --pandaproxy-addr internal://broker:8082
+ - --advertise-pandaproxy-addr internal://broker:8082
+ - --schema-registry-addr internal://0.0.0.0:8081
+ - --rpc-addr broker:33145
+ - --advertise-rpc-addr broker:33145
+ - --mode dev-container
+ - --default-log-level=warn
+ healthcheck:
+ test: ["CMD-SHELL", "rpk cluster health | grep -E 'Healthy:.+true' || exit 1"]
+ interval: 30s
+ timeout: 10s
+ retries: 5
+ start_period: 10s
+
+ main_router_service:
+ image: docker.io/library/main-router:latest
+ container_name: main_router_service
+ ports:
+ - "8082:8082"
+ networks:
+ - test-dynamic-router
+ environment:
+ - THC_PATH=/main-router/actuator/health
+ - THC_PORT=8082
+ healthcheck:
+ test: [ "CMD", "/cnb/process/health-check" ]
+ interval: 30s
+ timeout: 10s
+ retries: 5
+ start_period: 10s
+ depends_on:
+ redpanda_broker:
+ condition: service_healthy
+
+ all_numbers_service:
+ image: docker.io/library/all-numbers-service:latest
+ container_name: all_numbers_service
+ ports:
+ - "8911:8911"
+ networks:
+ - test-dynamic-router
+ environment:
+ - THC_PATH=/all-numbers/actuator/health
+ - THC_PORT=8911
+ healthcheck:
+ test: [ "CMD", "/cnb/process/health-check" ]
+ interval: 30s
+ timeout: 10s
+ retries: 5
+ start_period: 10s
+ depends_on:
+ main_router_service:
+ condition: service_healthy
+
+ even_numbers_service:
+ image: docker.io/library/even-numbers-service:latest
+ container_name: even_numbers_service
+ ports:
+ - "8902:8902"
+ networks:
+ - test-dynamic-router
+ environment:
+ - THC_PATH=/even-numbers/actuator/health
+ - THC_PORT=8902
+ healthcheck:
+ test: [ "CMD", "/cnb/process/health-check" ]
+ interval: 30s
+ timeout: 10s
+ retries: 5
+ start_period: 10s
+ depends_on:
+ main_router_service:
+ condition: service_healthy
+
+ odd_numbers_service:
+ image: docker.io/library/odd-numbers-service:latest
+ container_name: odd_numbers_service
+ ports:
+ - "8901:8901"
+ networks:
+ - test-dynamic-router
+ environment:
+ - THC_PATH=/odd-numbers/actuator/health
+ - THC_PORT=8901
+ healthcheck:
+ test: [ "CMD", "/cnb/process/health-check" ]
+ interval: 30s
+ timeout: 10s
+ retries: 5
+ start_period: 10s
+ depends_on:
+ main_router_service:
+ condition: service_healthy
+
+networks:
+ test-dynamic-router:
+ driver: bridge
+ ipam:
+ config:
+ - subnet: 10.5.0.0/16
+
+volumes:
+ broker: null
\ No newline at end of file
diff --git a/dynamic-router-eip/pom.xml b/dynamic-router-eip/pom.xml
index 58f27bc..5a879c3 100644
--- a/dynamic-router-eip/pom.xml
+++ b/dynamic-router-eip/pom.xml
@@ -26,11 +26,19 @@
</parent>
<artifactId>camel-example-spring-boot-dynamic-router-eip</artifactId>
- <name>Camel SB Examples :: Dynamic Router EIP</name>
- <description>An example on how to use the Dynamic Router EIP component in Spring Boot</description>
+ <packaging>pom</packaging>
+
+ <name>Camel SB Examples :: Dynamic Router EIP :: Examples</name>
+ <description>Dynamic Router EIP component examples</description>
+
+ <modules>
+ <module>dynamic-router-eip-single</module>
+ <module>dynamic-router-eip-stack</module>
+ </modules>
<properties>
<category>EIP</category>
+ <spring-statemachine.version>4.0.0</spring-statemachine.version>
</properties>
<!-- Spring-Boot and Camel BOM -->
@@ -50,77 +58,131 @@
<type>pom</type>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>org.springframework.statemachine</groupId>
+ <artifactId>spring-statemachine-bom</artifactId>
+ <version>${spring-statemachine.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
<dependencies>
- <!-- Camel -->
<dependency>
- <groupId>org.apache.camel.springboot</groupId>
- <artifactId>camel-dynamic-router-starter</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.camel.springboot</groupId>
- <artifactId>camel-direct-starter</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.camel.springboot</groupId>
- <artifactId>camel-bean-starter</artifactId>
- <version>${project.version}</version>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-configuration-processor</artifactId>
- <optional>true</optional>
+ <artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
- <groupId>org.junit.jupiter</groupId>
- <artifactId>junit-jupiter-engine</artifactId>
- <scope>test</scope>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-spring-boot-starter</artifactId>
</dependency>
<dependency>
- <groupId>org.junit.platform</groupId>
- <artifactId>junit-platform-launcher</artifactId>
- <scope>test</scope>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-spring-starter</artifactId>
</dependency>
<dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-test</artifactId>
- <scope>test</scope>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-dynamic-router-starter</artifactId>
</dependency>
</dependencies>
<build>
+ <pluginManagement>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <version>${spring-boot-version}</version>
+ <configuration combine.children="append">
+ <image>
+ <name>docker.io/library/${project.artifactId}</name>
+ <env>
+ <BPE_DELIM_JAVA_TOOL_OPTIONS xml:space="preserve"> </BPE_DELIM_JAVA_TOOL_OPTIONS>
+ <BPE_APPEND_JAVA_TOOL_OPTIONS>--add-opens=java.base/sun.net=ALL-UNNAMED</BPE_APPEND_JAVA_TOOL_OPTIONS>
+ <BP_HEALTH_CHECKER_ENABLED>true</BP_HEALTH_CHECKER_ENABLED>
+ <THC_PATH>/actuator/health</THC_PATH>
+ </env>
+ <buildpacks>
+ <buildpack>urn:cnb:builder:paketo-buildpacks/java</buildpack>
+ <buildpack>gcr.io/paketo-buildpacks/health-checker:latest</buildpack>
+ </buildpacks>
+ </image>
+ <layers>
+ <enabled>true</enabled>
+ </layers>
+ <jvmArguments>--add-opens=java.base/sun.net=ALL-UNNAMED</jvmArguments>
+ </configuration>
+ </plugin>
+ </plugins>
+ </pluginManagement>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
- <version>${spring-boot-version}</version>
<executions>
<execution>
+ <id>process-aot</id>
<goals>
- <goal>repackage</goal>
+ <goal>process-aot</goal>
</goals>
</execution>
</executions>
</plugin>
- <plugin>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-package-maven-plugin</artifactId>
- <version>${camel-version}</version>
- <executions>
- <execution>
- <id>generate</id>
- <goals>
- <goal>generate-component</goal>
- </goals>
- <phase>process-classes</phase>
- </execution>
- </executions>
- </plugin>
</plugins>
</build>
-
-</project>
+ <profiles>
+ <profile>
+ <id>podman</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <configuration>
+ <docker>
+ <host>${XDG_RUNTIME_DIR}/podman/podman.sock</host>
+ <bindHostToBuilder>true</bindHostToBuilder>
+ </docker>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>build-image-no-fork</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>docker</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <configuration>
+ <docker>
+ <host>/var/run/docker.sock</host>
+ <bindHostToBuilder>true</bindHostToBuilder>
+ </docker>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>build-image-no-fork</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+</project>
\ No newline at end of file
diff --git a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/AllRecipientsApplicationTest.java b/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/AllRecipientsApplicationTest.java
deleted file mode 100644
index 851a5a0..0000000
--- a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/AllRecipientsApplicationTest.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.apache.camel.example.springboot;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ActiveProfiles;
-
-@SpringBootTest(classes = { Application.class })
-@ActiveProfiles("test-all")
-class AllRecipientsApplicationTest {
-
- @Test
- void testApplication() {
- Assertions.assertTrue(true);
- }
-}
diff --git a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/FirstRecipientApplicationTest.java b/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/FirstRecipientApplicationTest.java
deleted file mode 100644
index c363ac9..0000000
--- a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/FirstRecipientApplicationTest.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.apache.camel.example.springboot;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ActiveProfiles;
-
-@SpringBootTest(classes = { Application.class })
-@ActiveProfiles("test-first")
-class FirstRecipientApplicationTest {
-
- @Test
- void testApplication() {
- Assertions.assertTrue(true);
- }
-}
diff --git a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/LessExpectedApplicationTest.java b/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/LessExpectedApplicationTest.java
deleted file mode 100644
index 2d889eb..0000000
--- a/dynamic-router-eip/src/test/java/org/apache/camel/example/springboot/LessExpectedApplicationTest.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.apache.camel.example.springboot;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ActiveProfiles;
-
-@SpringBootTest(classes = { Application.class })
-@ActiveProfiles("test-less")
-class LessExpectedApplicationTest {
-
- @Test
- void testApplication() {
- Assertions.assertTrue(true);
- }
-}
diff --git a/dynamic-router-eip/src/test/resources/logback-test.xml b/dynamic-router-eip/src/test/resources/logback-test.xml
deleted file mode 100644
index 2fb8398..0000000
--- a/dynamic-router-eip/src/test/resources/logback-test.xml
+++ /dev/null
@@ -1,14 +0,0 @@
-<configuration>
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>%-9relative [%-5p] %-30c{0} - %m%n</pattern>
- </encoder>
- </appender>
- <root level="error">
- <appender-ref ref="STDOUT"/>
- </root>
- <logger name="org.apache.camel.example.springboot.AllRecipientsApplicationTest" level="warn"/>
- <logger name="org.apache.camel.example.springboot.FirstRecipientApplicationTest" level="warn"/>
- <logger name="org.apache.camel.example.springboot.LessExpectedApplicationTest" level="warn"/>
- <logger name="org.apache.camel.example.springboot" level="info"/>
-</configuration>