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.git
The following commit(s) were added to refs/heads/main by this push:
new 010251049e09 CAMEL-23220: Added support for idempotent consumption in
the camel-at… (#22114)
010251049e09 is described below
commit 010251049e0929e63573b158a025897cbc6dab46
Author: Dennis Bohnstedt <[email protected]>
AuthorDate: Thu Apr 30 10:07:35 2026 +0200
CAMEL-23220: Added support for idempotent consumption in the camel-at…
(#22114)
* CAMEL-23220: Added support for idempotent consumption in the camel-atom
component.
The fix support idempotency with a plugable strategy for recognizing
already seen atom entries. Them component defaults to
idempotent consumption, as it did in versions up to 3.16.
* CAMEL-23220: Added camel-catalog file for Camel-Atom idempotency options
* CAMEL-23220: Added missing AtomEndpointBuilderFactory for Camel-Atom
idempotency change
* CAMEL-23220: Added @UriParam for idempotentRepository to enable pluggable
IdempotentRepository selection, and updated documentation of idempotency in the
compoent.
---
.../org/apache/camel/catalog/components/atom.json | 31 +-
.../component/atom/AtomEndpointConfigurer.java | 15 +
.../component/atom/AtomEndpointUriFactory.java | 5 +-
.../org/apache/camel/component/atom/atom.json | 31 +-
.../camel-atom/src/main/docs/atom-component.adoc | 38 +++
.../apache/camel/component/atom/AtomComponent.java | 45 ++-
.../apache/camel/component/atom/AtomEndpoint.java | 48 ++++
.../component/atom/AtomEntryPollingConsumer.java | 18 +-
.../component/atom/AtomIdempotentStrategy.java | 27 ++
.../atom/ItemUpdatedIdempotentStrategy.java | 66 +++++
.../atom/RepositoryGuidIdempotentStrategy.java | 58 ++++
.../component/feed/FeedEntryPollingConsumer.java | 16 +-
.../camel-atom/src/test/data/updated-feed.atom | 316 +++++++++++++++++++++
.../atom/AtomGoodBlogDefaultIdempotencyTest.java | 125 ++++++++
.../atom/AtomGoodBlogNotIdempotentTest.java | 76 +++++
...ava => AtomGoodBlogsCustomItempotencyTest.java} | 70 ++---
...sRepositoryItempotencyCustomRepositoryTest.java | 113 ++++++++
...=> AtomGoodBlogsRepositoryItempotencyTest.java} | 80 ++----
.../camel/component/atom/AtomGoodBlogsTest.java | 2 +-
.../atom/ItemUpdatedIdempotentStrategyTest.java | 117 ++++++++
.../atom/RepositoryGuidIdempotentStrategyTest.java | 68 +++++
.../endpoint/dsl/AtomEndpointBuilderFactory.java | 124 ++++++++
22 files changed, 1357 insertions(+), 132 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/atom.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/atom.json
index b72bfb770777..2d648aab5ee9 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/atom.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/atom.json
@@ -41,19 +41,22 @@
"exchangePattern": { "index": 7, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "index": 8, "kind": "parameter", "displayName": "Poll
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": false, "secret": false, "description": "A pluggable
org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your
custom implementation to control error handling usually occurred during the pol
[...]
"feedHeader": { "index": 9, "kind": "parameter", "displayName": "Feed
Header", "group": "advanced", "label": "advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "description": "Sets whether to add the
feed object as a header." },
- "backoffErrorThreshold": { "index": 10, "kind": "parameter",
"displayName": "Backoff Error Threshold", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "integer", "javaType": "int",
"deprecated": false, "autowired": false, "secret": false, "description": "The
number of subsequent error polls (failed due some error) that should happen
before the backoffMultipler should kick-in." },
- "backoffIdleThreshold": { "index": 11, "kind": "parameter", "displayName":
"Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "The number of subsequent
idle polls that should happen before the backoffMultipler should kick-in." },
- "backoffMultiplier": { "index": 12, "kind": "parameter", "displayName":
"Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "To let the scheduled
polling consumer backoff if there has been a number of subsequent idles\/errors
in a row. The multiplier is then the number of polls that will be skipped
before the next actual attempt is h [...]
- "delay": { "index": 13, "kind": "parameter", "displayName": "Delay",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 500, "description": "Milliseconds before the
next poll." },
- "greedy": { "index": 14, "kind": "parameter", "displayName": "Greedy",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If greedy is enabled,
then the ScheduledPollConsumer will run immediately again, if the previous run
polled 1 or more messages." },
- "initialDelay": { "index": 15, "kind": "parameter", "displayName":
"Initial Delay", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000, "description":
"Milliseconds before the first poll starts." },
- "repeatCount": { "index": 16, "kind": "parameter", "displayName": "Repeat
Count", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 0, "description": "Specifies a maximum limit
of number of fires. So if you set it to 1, the scheduler will only fire once.
If you set it to 5, it will only fire five times. A value of zero or negative
means fire forever." },
- "runLoggingLevel": { "index": 17, "kind": "parameter", "displayName": "Run
Logging Level", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "enum", "javaType": "org.apache.camel.LoggingLevel",
"enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "TRACE",
"description": "The consumer logs a start\/complete log line when it polls.
This option allows you to configure the log [...]
- "scheduledExecutorService": { "index": 18, "kind": "parameter",
"displayName": "Scheduled Executor Service", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "object", "javaType":
"java.util.concurrent.ScheduledExecutorService", "deprecated": false,
"autowired": false, "secret": false, "description": "Allows for configuring a
custom\/shared thread pool to use for the consumer. By default each consumer
has its own single threaded thread pool." },
- "scheduler": { "index": 19, "kind": "parameter", "displayName":
"Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "object", "javaType": "java.lang.Object", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "none", "description": "To
use a cron scheduler from either camel-spring or camel-quartz component. Use
value spring or quartz for built in scheduler" },
- "schedulerProperties": { "index": 20, "kind": "parameter", "displayName":
"Scheduler Properties", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "object", "javaType":
"java.util.Map<java.lang.String, java.lang.Object>", "prefix": "scheduler.",
"multiValue": true, "deprecated": false, "autowired": false, "secret": false,
"description": "To configure additional properties when using a custom
scheduler or any of the Quartz, Spring based scheduler. This i [...]
- "startScheduler": { "index": 21, "kind": "parameter", "displayName":
"Start Scheduler", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"description": "Whether the scheduler should be auto started." },
- "timeUnit": { "index": 22, "kind": "parameter", "displayName": "Time
Unit", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "enum", "javaType": "java.util.concurrent.TimeUnit", "enum": [
"NANOSECONDS", "MICROSECONDS", "MILLISECONDS", "SECONDS", "MINUTES", "HOURS",
"DAYS" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and
delay options." },
- "useFixedDelay": { "index": 23, "kind": "parameter", "displayName": "Use
Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in
JDK for details." }
+ "idempotent": { "index": 10, "kind": "parameter", "displayName":
"Idempotent", "group": "filter", "label": "consumer,filter", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "description": "Option to use the
Idempotent Consumer EIP pattern to let Camel skip already processed entries. By
default it skips Atom entries that is older that the last read entry, using
Atom updatedDate and publishedD [...]
+ "idempotentRepository": { "index": 11, "kind": "parameter", "displayName":
"Idempotent Repository", "group": "filter", "label": "consumer,filter",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.IdempotentRepository", "deprecated": false, "autowired":
false, "secret": false, "description": "A pluggable repository
org.apache.camel.spi.IdempotentRepository which by default use
MemoryIdempotentRepository if none is specified and idempotent is true." },
+ "idempotentStrategy": { "index": 12, "kind": "parameter", "displayName":
"Idempotent Strategy", "group": "filter (advanced)", "label":
"consumer,filter,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.component.atom.AtomIdempotentStrategy", "enum": [ "default",
"repository" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "default", "description": "A pluggable strategy
org.apache.camel.component.FeedIdempotentStrategy to use when [...]
+ "backoffErrorThreshold": { "index": 13, "kind": "parameter",
"displayName": "Backoff Error Threshold", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "integer", "javaType": "int",
"deprecated": false, "autowired": false, "secret": false, "description": "The
number of subsequent error polls (failed due some error) that should happen
before the backoffMultipler should kick-in." },
+ "backoffIdleThreshold": { "index": 14, "kind": "parameter", "displayName":
"Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "The number of subsequent
idle polls that should happen before the backoffMultipler should kick-in." },
+ "backoffMultiplier": { "index": 15, "kind": "parameter", "displayName":
"Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "To let the scheduled
polling consumer backoff if there has been a number of subsequent idles\/errors
in a row. The multiplier is then the number of polls that will be skipped
before the next actual attempt is h [...]
+ "delay": { "index": 16, "kind": "parameter", "displayName": "Delay",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 500, "description": "Milliseconds before the
next poll." },
+ "greedy": { "index": 17, "kind": "parameter", "displayName": "Greedy",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If greedy is enabled,
then the ScheduledPollConsumer will run immediately again, if the previous run
polled 1 or more messages." },
+ "initialDelay": { "index": 18, "kind": "parameter", "displayName":
"Initial Delay", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000, "description":
"Milliseconds before the first poll starts." },
+ "repeatCount": { "index": 19, "kind": "parameter", "displayName": "Repeat
Count", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 0, "description": "Specifies a maximum limit
of number of fires. So if you set it to 1, the scheduler will only fire once.
If you set it to 5, it will only fire five times. A value of zero or negative
means fire forever." },
+ "runLoggingLevel": { "index": 20, "kind": "parameter", "displayName": "Run
Logging Level", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "enum", "javaType": "org.apache.camel.LoggingLevel",
"enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "TRACE",
"description": "The consumer logs a start\/complete log line when it polls.
This option allows you to configure the log [...]
+ "scheduledExecutorService": { "index": 21, "kind": "parameter",
"displayName": "Scheduled Executor Service", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "object", "javaType":
"java.util.concurrent.ScheduledExecutorService", "deprecated": false,
"autowired": false, "secret": false, "description": "Allows for configuring a
custom\/shared thread pool to use for the consumer. By default each consumer
has its own single threaded thread pool." },
+ "scheduler": { "index": 22, "kind": "parameter", "displayName":
"Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "object", "javaType": "java.lang.Object", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "none", "description": "To
use a cron scheduler from either camel-spring or camel-quartz component. Use
value spring or quartz for built in scheduler" },
+ "schedulerProperties": { "index": 23, "kind": "parameter", "displayName":
"Scheduler Properties", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "object", "javaType":
"java.util.Map<java.lang.String, java.lang.Object>", "prefix": "scheduler.",
"multiValue": true, "deprecated": false, "autowired": false, "secret": false,
"description": "To configure additional properties when using a custom
scheduler or any of the Quartz, Spring based scheduler. This i [...]
+ "startScheduler": { "index": 24, "kind": "parameter", "displayName":
"Start Scheduler", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"description": "Whether the scheduler should be auto started." },
+ "timeUnit": { "index": 25, "kind": "parameter", "displayName": "Time
Unit", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "enum", "javaType": "java.util.concurrent.TimeUnit", "enum": [
"NANOSECONDS", "MICROSECONDS", "MILLISECONDS", "SECONDS", "MINUTES", "HOURS",
"DAYS" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and
delay options." },
+ "useFixedDelay": { "index": 26, "kind": "parameter", "displayName": "Use
Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in
JDK for details." }
}
}
diff --git
a/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointConfigurer.java
b/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointConfigurer.java
index b99264ba8ca6..5a8ebb8798f2 100644
---
a/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointConfigurer.java
+++
b/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointConfigurer.java
@@ -39,6 +39,11 @@ public class AtomEndpointConfigurer extends
PropertyConfigurerSupport implements
case "feedheader":
case "feedHeader": target.setFeedHeader(property(camelContext,
boolean.class, value)); return true;
case "greedy": target.setGreedy(property(camelContext, boolean.class,
value)); return true;
+ case "idempotent": target.setIdempotent(property(camelContext,
boolean.class, value)); return true;
+ case "idempotentrepository":
+ case "idempotentRepository":
target.setIdempotentRepository(property(camelContext,
org.apache.camel.spi.IdempotentRepository.class, value)); return true;
+ case "idempotentstrategy":
+ case "idempotentStrategy":
target.setIdempotentStrategy(property(camelContext,
org.apache.camel.component.atom.AtomIdempotentStrategy.class, value)); return
true;
case "initialdelay":
case "initialDelay": target.setInitialDelay(property(camelContext,
long.class, value)); return true;
case "pollstrategy":
@@ -89,6 +94,11 @@ public class AtomEndpointConfigurer extends
PropertyConfigurerSupport implements
case "feedheader":
case "feedHeader": return boolean.class;
case "greedy": return boolean.class;
+ case "idempotent": return boolean.class;
+ case "idempotentrepository":
+ case "idempotentRepository": return
org.apache.camel.spi.IdempotentRepository.class;
+ case "idempotentstrategy":
+ case "idempotentStrategy": return
org.apache.camel.component.atom.AtomIdempotentStrategy.class;
case "initialdelay":
case "initialDelay": return long.class;
case "pollstrategy":
@@ -140,6 +150,11 @@ public class AtomEndpointConfigurer extends
PropertyConfigurerSupport implements
case "feedheader":
case "feedHeader": return target.isFeedHeader();
case "greedy": return target.isGreedy();
+ case "idempotent": return target.isIdempotent();
+ case "idempotentrepository":
+ case "idempotentRepository": return target.getIdempotentRepository();
+ case "idempotentstrategy":
+ case "idempotentStrategy": return target.getIdempotentStrategy();
case "initialdelay":
case "initialDelay": return target.getInitialDelay();
case "pollstrategy":
diff --git
a/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointUriFactory.java
b/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointUriFactory.java
index d44ac38da116..4b62c6fcb83d 100644
---
a/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointUriFactory.java
+++
b/components/camel-atom/src/generated/java/org/apache/camel/component/atom/AtomEndpointUriFactory.java
@@ -23,7 +23,7 @@ public class AtomEndpointUriFactory extends
org.apache.camel.support.component.E
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(24);
+ Set<String> props = new HashSet<>(27);
props.add("backoffErrorThreshold");
props.add("backoffIdleThreshold");
props.add("backoffMultiplier");
@@ -34,6 +34,9 @@ public class AtomEndpointUriFactory extends
org.apache.camel.support.component.E
props.add("feedHeader");
props.add("feedUri");
props.add("greedy");
+ props.add("idempotent");
+ props.add("idempotentRepository");
+ props.add("idempotentStrategy");
props.add("initialDelay");
props.add("pollStrategy");
props.add("repeatCount");
diff --git
a/components/camel-atom/src/generated/resources/META-INF/org/apache/camel/component/atom/atom.json
b/components/camel-atom/src/generated/resources/META-INF/org/apache/camel/component/atom/atom.json
index b72bfb770777..2d648aab5ee9 100644
---
a/components/camel-atom/src/generated/resources/META-INF/org/apache/camel/component/atom/atom.json
+++
b/components/camel-atom/src/generated/resources/META-INF/org/apache/camel/component/atom/atom.json
@@ -41,19 +41,22 @@
"exchangePattern": { "index": 7, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "index": 8, "kind": "parameter", "displayName": "Poll
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": false, "secret": false, "description": "A pluggable
org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your
custom implementation to control error handling usually occurred during the pol
[...]
"feedHeader": { "index": 9, "kind": "parameter", "displayName": "Feed
Header", "group": "advanced", "label": "advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "description": "Sets whether to add the
feed object as a header." },
- "backoffErrorThreshold": { "index": 10, "kind": "parameter",
"displayName": "Backoff Error Threshold", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "integer", "javaType": "int",
"deprecated": false, "autowired": false, "secret": false, "description": "The
number of subsequent error polls (failed due some error) that should happen
before the backoffMultipler should kick-in." },
- "backoffIdleThreshold": { "index": 11, "kind": "parameter", "displayName":
"Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "The number of subsequent
idle polls that should happen before the backoffMultipler should kick-in." },
- "backoffMultiplier": { "index": 12, "kind": "parameter", "displayName":
"Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "To let the scheduled
polling consumer backoff if there has been a number of subsequent idles\/errors
in a row. The multiplier is then the number of polls that will be skipped
before the next actual attempt is h [...]
- "delay": { "index": 13, "kind": "parameter", "displayName": "Delay",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 500, "description": "Milliseconds before the
next poll." },
- "greedy": { "index": 14, "kind": "parameter", "displayName": "Greedy",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If greedy is enabled,
then the ScheduledPollConsumer will run immediately again, if the previous run
polled 1 or more messages." },
- "initialDelay": { "index": 15, "kind": "parameter", "displayName":
"Initial Delay", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000, "description":
"Milliseconds before the first poll starts." },
- "repeatCount": { "index": 16, "kind": "parameter", "displayName": "Repeat
Count", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 0, "description": "Specifies a maximum limit
of number of fires. So if you set it to 1, the scheduler will only fire once.
If you set it to 5, it will only fire five times. A value of zero or negative
means fire forever." },
- "runLoggingLevel": { "index": 17, "kind": "parameter", "displayName": "Run
Logging Level", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "enum", "javaType": "org.apache.camel.LoggingLevel",
"enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "TRACE",
"description": "The consumer logs a start\/complete log line when it polls.
This option allows you to configure the log [...]
- "scheduledExecutorService": { "index": 18, "kind": "parameter",
"displayName": "Scheduled Executor Service", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "object", "javaType":
"java.util.concurrent.ScheduledExecutorService", "deprecated": false,
"autowired": false, "secret": false, "description": "Allows for configuring a
custom\/shared thread pool to use for the consumer. By default each consumer
has its own single threaded thread pool." },
- "scheduler": { "index": 19, "kind": "parameter", "displayName":
"Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "object", "javaType": "java.lang.Object", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "none", "description": "To
use a cron scheduler from either camel-spring or camel-quartz component. Use
value spring or quartz for built in scheduler" },
- "schedulerProperties": { "index": 20, "kind": "parameter", "displayName":
"Scheduler Properties", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "object", "javaType":
"java.util.Map<java.lang.String, java.lang.Object>", "prefix": "scheduler.",
"multiValue": true, "deprecated": false, "autowired": false, "secret": false,
"description": "To configure additional properties when using a custom
scheduler or any of the Quartz, Spring based scheduler. This i [...]
- "startScheduler": { "index": 21, "kind": "parameter", "displayName":
"Start Scheduler", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"description": "Whether the scheduler should be auto started." },
- "timeUnit": { "index": 22, "kind": "parameter", "displayName": "Time
Unit", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "enum", "javaType": "java.util.concurrent.TimeUnit", "enum": [
"NANOSECONDS", "MICROSECONDS", "MILLISECONDS", "SECONDS", "MINUTES", "HOURS",
"DAYS" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and
delay options." },
- "useFixedDelay": { "index": 23, "kind": "parameter", "displayName": "Use
Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in
JDK for details." }
+ "idempotent": { "index": 10, "kind": "parameter", "displayName":
"Idempotent", "group": "filter", "label": "consumer,filter", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "description": "Option to use the
Idempotent Consumer EIP pattern to let Camel skip already processed entries. By
default it skips Atom entries that is older that the last read entry, using
Atom updatedDate and publishedD [...]
+ "idempotentRepository": { "index": 11, "kind": "parameter", "displayName":
"Idempotent Repository", "group": "filter", "label": "consumer,filter",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.IdempotentRepository", "deprecated": false, "autowired":
false, "secret": false, "description": "A pluggable repository
org.apache.camel.spi.IdempotentRepository which by default use
MemoryIdempotentRepository if none is specified and idempotent is true." },
+ "idempotentStrategy": { "index": 12, "kind": "parameter", "displayName":
"Idempotent Strategy", "group": "filter (advanced)", "label":
"consumer,filter,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.component.atom.AtomIdempotentStrategy", "enum": [ "default",
"repository" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "default", "description": "A pluggable strategy
org.apache.camel.component.FeedIdempotentStrategy to use when [...]
+ "backoffErrorThreshold": { "index": 13, "kind": "parameter",
"displayName": "Backoff Error Threshold", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "integer", "javaType": "int",
"deprecated": false, "autowired": false, "secret": false, "description": "The
number of subsequent error polls (failed due some error) that should happen
before the backoffMultipler should kick-in." },
+ "backoffIdleThreshold": { "index": 14, "kind": "parameter", "displayName":
"Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "The number of subsequent
idle polls that should happen before the backoffMultipler should kick-in." },
+ "backoffMultiplier": { "index": 15, "kind": "parameter", "displayName":
"Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "description": "To let the scheduled
polling consumer backoff if there has been a number of subsequent idles\/errors
in a row. The multiplier is then the number of polls that will be skipped
before the next actual attempt is h [...]
+ "delay": { "index": 16, "kind": "parameter", "displayName": "Delay",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 500, "description": "Milliseconds before the
next poll." },
+ "greedy": { "index": 17, "kind": "parameter", "displayName": "Greedy",
"group": "scheduler", "label": "consumer,scheduler", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If greedy is enabled,
then the ScheduledPollConsumer will run immediately again, if the previous run
polled 1 or more messages." },
+ "initialDelay": { "index": 18, "kind": "parameter", "displayName":
"Initial Delay", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "integer", "javaType": "long", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000, "description":
"Milliseconds before the first poll starts." },
+ "repeatCount": { "index": 19, "kind": "parameter", "displayName": "Repeat
Count", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 0, "description": "Specifies a maximum limit
of number of fires. So if you set it to 1, the scheduler will only fire once.
If you set it to 5, it will only fire five times. A value of zero or negative
means fire forever." },
+ "runLoggingLevel": { "index": 20, "kind": "parameter", "displayName": "Run
Logging Level", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "enum", "javaType": "org.apache.camel.LoggingLevel",
"enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "TRACE",
"description": "The consumer logs a start\/complete log line when it polls.
This option allows you to configure the log [...]
+ "scheduledExecutorService": { "index": 21, "kind": "parameter",
"displayName": "Scheduled Executor Service", "group": "scheduler", "label":
"consumer,scheduler", "required": false, "type": "object", "javaType":
"java.util.concurrent.ScheduledExecutorService", "deprecated": false,
"autowired": false, "secret": false, "description": "Allows for configuring a
custom\/shared thread pool to use for the consumer. By default each consumer
has its own single threaded thread pool." },
+ "scheduler": { "index": 22, "kind": "parameter", "displayName":
"Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "object", "javaType": "java.lang.Object", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "none", "description": "To
use a cron scheduler from either camel-spring or camel-quartz component. Use
value spring or quartz for built in scheduler" },
+ "schedulerProperties": { "index": 23, "kind": "parameter", "displayName":
"Scheduler Properties", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "object", "javaType":
"java.util.Map<java.lang.String, java.lang.Object>", "prefix": "scheduler.",
"multiValue": true, "deprecated": false, "autowired": false, "secret": false,
"description": "To configure additional properties when using a custom
scheduler or any of the Quartz, Spring based scheduler. This i [...]
+ "startScheduler": { "index": 24, "kind": "parameter", "displayName":
"Start Scheduler", "group": "scheduler", "label": "consumer,scheduler",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": true,
"description": "Whether the scheduler should be auto started." },
+ "timeUnit": { "index": 25, "kind": "parameter", "displayName": "Time
Unit", "group": "scheduler", "label": "consumer,scheduler", "required": false,
"type": "enum", "javaType": "java.util.concurrent.TimeUnit", "enum": [
"NANOSECONDS", "MICROSECONDS", "MILLISECONDS", "SECONDS", "MINUTES", "HOURS",
"DAYS" ], "deprecated": false, "autowired": false, "secret": false,
"defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and
delay options." },
+ "useFixedDelay": { "index": 26, "kind": "parameter", "displayName": "Use
Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in
JDK for details." }
}
}
diff --git a/components/camel-atom/src/main/docs/atom-component.adoc
b/components/camel-atom/src/main/docs/atom-component.adoc
index c951654dbfcb..72eaba3b6309 100644
--- a/components/camel-atom/src/main/docs/atom-component.adoc
+++ b/components/camel-atom/src/main/docs/atom-component.adoc
@@ -69,6 +69,44 @@ Depending on the `splitEntries` flag Camel will either
return one
Camel can set the `Feed` object on the In header (see `feedHeader`
option to disable this):
+== Usage
+
+=== Avoid reading the same entry more than once (idempotent consumer)
+
+Camel supports Idempotent Consumer
+directly within the component, so it will skip already processed entries.
+This feature is enabled by default, but can be disabled by setting the
`idempotent=false` option.
+
+[source,java]
+----
+from("file://inbox?idempotent=false").to("...");
+----
+
+The default Idempotency strategy uses Atom published and updated timestamps to
identify already
+read entries, by filtering all entries older that the _last read_ entry.
+
+The idempotency strategy can be changed by setting the `idempotentStrategy`
option. The components support two
+strategies _out-of-the-box_, but custom strategies can be used, by referencing
a subclass
+of `org.apache.camel.component.atom.AtomIdempotentStrategy`, with Apache
Camel's # notation.
+
+[source,java]
+----
+from("file://inbox?idempotentStrategy=#myIdempotentStrategy").to("...");
+----
+
+==== Repository idempotent strategy
+As an alternative to the default idempotent strategy, a repository based
strategy can be use:
+
+[source,java]
+----
+from("file://inbox?idempotentStrategy=repository").to("...");
+----
+
+This strategy checks for idempotency by storing the Atom Entry GUID in an
`IdempotentRepository`. The strategy defaults
+a MemoryIdempotentRepository, but the repository can be changes to any
subclass of `IdempotentRepository`, using the
+`idempotentRepository` option.
+
+
== Examples
=== Consumer Example
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomComponent.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomComponent.java
index 603cc48b3977..fb59104df67e 100644
---
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomComponent.java
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomComponent.java
@@ -22,6 +22,7 @@ import java.util.Map;
import org.apache.camel.Endpoint;
import org.apache.camel.component.feed.FeedComponent;
import org.apache.camel.component.feed.FeedEndpoint;
+import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.util.URISupport;
@@ -31,12 +32,34 @@ import org.apache.camel.util.URISupport;
@Component("atom")
public class AtomComponent extends FeedComponent {
+ private static final String KEY_FORMAT_IDEMPOTENT_PARAM = "idempotent";
+ private static final String KEY_FORMAT_STRATEGY_PARAM =
"idempotentStrategy";
+ private static final String KEY_FORMAT_IDEMPOTENT_REPOSITORY_PARAM =
"idempotentRepository";
+
public AtomComponent() {
}
@Override
protected FeedEndpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
- return new AtomEndpoint(uri, this, null);
+ AtomEndpoint endpoint = new AtomEndpoint(uri, this, null);
+ // Setup idempotency
+ endpoint.setIdempotent(getAndRemoveParameter(parameters,
KEY_FORMAT_IDEMPOTENT_PARAM, Boolean.class, true));
+ if (endpoint.isIdempotent()) {
+ endpoint.setIdempotentRepository(
+ resolveAndRemoveReferenceParameter(parameters,
KEY_FORMAT_IDEMPOTENT_REPOSITORY_PARAM,
+ IdempotentRepository.class));
+ String idempotentStrategy = getAndRemoveParameter(parameters,
KEY_FORMAT_STRATEGY_PARAM, String.class);
+ AtomIdempotentStrategy atomIdempotentStrategy
+ = resolveEnumIdempotentStrategy(idempotentStrategy,
endpoint.getIdempotentRepository());
+ if (atomIdempotentStrategy != null) {
+ parameters.put(KEY_FORMAT_STRATEGY_PARAM,
atomIdempotentStrategy);
+ } else if (idempotentStrategy != null) {
+ // its not a standard, but a reference - fallback to
AtomEndpointConfigurator
+ parameters.put(KEY_FORMAT_STRATEGY_PARAM, idempotentStrategy);
+
//endpoint.setIdempotentStrategy(resolveAndRemoveReferenceParameter(parameters,
KEY_FORMAT_STRATEGY_PARAM, AtomIdempotentStrategy.class));
+ }
+ }
+ return endpoint;
}
@Override
@@ -61,4 +84,24 @@ public class AtomComponent extends FeedComponent {
atom.setFeedUri(feedUri);
}
+ /**
+ * Resolves the standard supported {@link AtomIdempotentStrategy} by a
name which can be:
+ * <ul>
+ * <li>default - to use the default date strategy</li>
+ * <li>repository - to use the Guid Repository strategy</li>
+ * </ul>
+ *
+ * @param name the name
+ * @return the strategy, or <tt>null</tt> if not a standard name.
+ */
+ private static AtomIdempotentStrategy resolveEnumIdempotentStrategy(
+ String name, IdempotentRepository idempotentRepository) {
+ if ("default".equalsIgnoreCase(name)) {
+ return new ItemUpdatedIdempotentStrategy();
+ } else if ("repository".equalsIgnoreCase(name)) {
+ return new RepositoryGuidIdempotentStrategy(idempotentRepository);
+ } else {
+ return null;
+ }
+ }
}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEndpoint.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEndpoint.java
index f54b66c88a30..1806f1549f82 100644
---
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEndpoint.java
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEndpoint.java
@@ -23,7 +23,9 @@ import org.apache.camel.component.feed.FeedComponent;
import org.apache.camel.component.feed.FeedEndpoint;
import org.apache.camel.component.feed.FeedPollingConsumer;
import org.apache.camel.spi.EndpointServiceLocation;
+import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
/**
* Poll Atom RSS feeds.
@@ -32,6 +34,22 @@ import org.apache.camel.spi.UriEndpoint;
category = { Category.DOCUMENT }, lenientProperties = true,
headersClass = AtomConstants.class)
public class AtomEndpoint extends FeedEndpoint implements
EndpointServiceLocation {
+ @UriParam(label = "consumer,filter", defaultValue = "true", description =
"Option to use the Idempotent "
+
+ "Consumer EIP pattern to let Camel skip already processed entries. By default
it skips Atom entries "
+
+ "that is older that the last read entry, using Atom updatedDate and
publishedDate, but this strategy "
+
+ "can be changed with the idempotentStrategy option. Idempotency only works
when splitEntries = true.")
+ protected boolean idempotent = true;
+ @UriParam(label = "consumer,filter,advanced", defaultValue = "default",
enums = "default,repository",
+ description = "A pluggable strategy
org.apache.camel.component.FeedIdempotentStrategy "
+ + "to use when checking idempotency. Camel
provides two implementations out of the box: default and repository. The
updated strategy is used as default if idempotent = true. "
+ + "The default strategy checks the Atom entrys
updated or published date is newer than the previously read entry. "
+ + "You can provide your own implementation of the
org.apache.camel.component.FeedIdempotentStrategy and refer to it using the #
notation.")
+ protected AtomIdempotentStrategy idempotentStrategy = new
ItemUpdatedIdempotentStrategy();
+
+ @UriParam(label = "consumer,filter", description = "A pluggable repository
org.apache.camel.spi.IdempotentRepository "
+ + "which by default use
MemoryIdempotentRepository if none is specified and idempotent is true.")
+ protected IdempotentRepository idempotentRepository;
+
public AtomEndpoint() {
}
@@ -78,4 +96,34 @@ public class AtomEndpoint extends FeedEndpoint implements
EndpointServiceLocatio
configureConsumer(answer);
return answer;
}
+
+ public boolean isIdempotent() {
+ return idempotent;
+ }
+
+ public FeedEndpoint setIdempotent(Boolean idempotent) {
+ this.idempotent = idempotent;
+ return this;
+ }
+
+ public AtomIdempotentStrategy getIdempotentStrategy() {
+ return idempotentStrategy;
+ }
+
+ public FeedEndpoint setIdempotentStrategy(AtomIdempotentStrategy
idempotentStrategy) {
+ this.idempotentStrategy = idempotentStrategy;
+ return this;
+ }
+
+ public IdempotentRepository getIdempotentRepository() {
+ return idempotentRepository;
+ }
+
+ /**
+ * A pluggable repository org.apache.camel.spi.IdempotentRepository which
by default use MemoryIdempotentRepository
+ * if none is specified and idempotent is true.
+ */
+ public void setIdempotentRepository(IdempotentRepository
idempotentRepository) {
+ this.idempotentRepository = idempotentRepository;
+ }
}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEntryPollingConsumer.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEntryPollingConsumer.java
index 311f26b1e007..811503e9a03a 100644
---
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEntryPollingConsumer.java
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomEntryPollingConsumer.java
@@ -27,7 +27,7 @@ import
org.apache.camel.component.feed.FeedEntryPollingConsumer;
/**
* Consumer to poll atom feeds and return each entry from the feed step by
step.
*/
-public class AtomEntryPollingConsumer extends FeedEntryPollingConsumer {
+public class AtomEntryPollingConsumer extends FeedEntryPollingConsumer<Item> {
private RssReader rssReader;
private List<Item> items;
@@ -36,6 +36,11 @@ public class AtomEntryPollingConsumer extends
FeedEntryPollingConsumer {
super(endpoint, processor, throttleEntries);
}
+ @Override
+ public AtomEndpoint getEndpoint() {
+ return (AtomEndpoint) super.getEndpoint();
+ }
+
@Override
protected void doStart() throws Exception {
this.rssReader = new RssReader();
@@ -60,11 +65,20 @@ public class AtomEntryPollingConsumer extends
FeedEntryPollingConsumer {
private List<Item> readItems() throws IOException {
if (items == null) {
- items = AtomUtils.readItems(endpoint.getCamelContext(),
endpoint.getFeedUri(), rssReader, endpoint.isSortEntries());
+ items = AtomUtils.readItems(getEndpoint().getCamelContext(),
getEndpoint().getFeedUri(), rssReader,
+ getEndpoint().isSortEntries());
list = items;
entryIndex = list.size() - 1;
}
return items;
}
+ @Override
+ protected boolean isValidItem(Item entry) {
+ boolean valid = true;
+ if (getEndpoint().isIdempotent()) {
+ valid = getEndpoint().getIdempotentStrategy().isValidItem(entry);
+ }
+ return valid;
+ }
}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomIdempotentStrategy.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomIdempotentStrategy.java
new file mode 100644
index 000000000000..072f6b9790f2
--- /dev/null
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/AtomIdempotentStrategy.java
@@ -0,0 +1,27 @@
+/*
+ * 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.component.atom;
+
+import com.apptasticsoftware.rssreader.Item;
+
+/**
+ *
+ */
+public interface AtomIdempotentStrategy {
+
+ boolean isValidItem(Item item);
+}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategy.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategy.java
new file mode 100644
index 000000000000..06040eb70ad3
--- /dev/null
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategy.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.atom;
+
+import java.text.ParseException;
+import java.util.Date;
+import java.util.Optional;
+
+import com.apptasticsoftware.rssreader.Item;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ */
+public class ItemUpdatedIdempotentStrategy implements AtomIdempotentStrategy {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ItemUpdatedIdempotentStrategy.class);
+ private Date lastUpdate;
+
+ @Override
+ public boolean isValidItem(Item item) {
+ Date updated = toDate(item.getUpdated());
+ if (updated == null) {
+ // never been updated so get published date
+ updated = toDate(item.getPubDate());
+ }
+ if (updated == null) {
+ LOG.debug("No updated time for entry so assuming its valid:
entry=[{}]", item);
+ return true;
+ }
+ if (lastUpdate != null) {
+ // we need to skip the latest updated entry
+ if (lastUpdate.after(updated) || lastUpdate.equals(updated)) {
+ LOG.debug("Entry is older than lastupdate=[{}], no valid
entry=[{}]", lastUpdate, item);
+ return false;
+ }
+ }
+ lastUpdate = updated;
+ return true;
+ }
+
+ private Date toDate(Optional<String> dateString) {
+ try {
+ return AtomConverter.toDate(dateString.orElse(null));
+ } catch (NullPointerException | ParseException e) {
+ LOG.debug("Failed to parse date string: {}", dateString, e);
+ return null;
+ }
+ }
+
+}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategy.java
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategy.java
new file mode 100644
index 000000000000..ef917f18f3af
--- /dev/null
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategy.java
@@ -0,0 +1,58 @@
+/*
+ * 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.component.atom;
+
+import com.apptasticsoftware.rssreader.Item;
+import org.apache.camel.spi.IdempotentRepository;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+
+/**
+ *
+ */
+public class RepositoryGuidIdempotentStrategy implements
AtomIdempotentStrategy {
+
+ IdempotentRepository idempotentRepository = null;
+
+ public RepositoryGuidIdempotentStrategy() {
+ this(null);
+ }
+
+ public RepositoryGuidIdempotentStrategy(IdempotentRepository
idempotentRepository) {
+ this.idempotentRepository = idempotentRepository;
+ }
+
+ @Override
+ public boolean isValidItem(Item entry) {
+ boolean valid = true;
+ if (entry != null && entry.getGuid().isPresent()) {
+ valid = getIdempotentRepository().add(entry.getGuid().get());
+ }
+ return valid;
+ }
+
+ protected IdempotentRepository getIdempotentRepository() {
+ if (this.idempotentRepository == null) {
+ this.idempotentRepository =
MemoryIdempotentRepository.memoryIdempotentRepository();
+ }
+ return this.idempotentRepository;
+ }
+
+ protected RepositoryGuidIdempotentStrategy
setIdempotentRepository(IdempotentRepository idempotentRepository) {
+ this.idempotentRepository = idempotentRepository;
+ return this;
+ }
+}
diff --git
a/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEntryPollingConsumer.java
b/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEntryPollingConsumer.java
index 91ead9052b5b..8fd76743643b 100644
---
a/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEntryPollingConsumer.java
+++
b/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEntryPollingConsumer.java
@@ -49,11 +49,13 @@ public abstract class FeedEntryPollingConsumer<E> extends
FeedPollingConsumer {
E entry = list.get(entryIndex--);
polledMessages++;
- Exchange exchange = endpoint.createExchange(feed, entry);
- getProcessor().process(exchange);
- if (this.throttleEntries) {
- // return and wait for the next poll to continue from last
time (this consumer is stateful)
- return polledMessages;
+ if (isValidItem(entry)) {
+ Exchange exchange = endpoint.createExchange(feed, entry);
+ getProcessor().process(exchange);
+ if (this.throttleEntries) {
+ // return and wait for the next poll to continue from last
time (this consumer is stateful)
+ return polledMessages;
+ }
}
}
@@ -64,6 +66,10 @@ public abstract class FeedEntryPollingConsumer<E> extends
FeedPollingConsumer {
return polledMessages;
}
+ protected boolean isValidItem(E entry) {
+ return true;
+ }
+
protected abstract void resetList();
protected abstract void populateList(Object feed) throws Exception;
diff --git a/components/camel-atom/src/test/data/updated-feed.atom
b/components/camel-atom/src/test/data/updated-feed.atom
new file mode 100644
index 000000000000..c3c6f9963d5c
--- /dev/null
+++ b/components/camel-atom/src/test/data/updated-feed.atom
@@ -0,0 +1,316 @@
+<?xml version='1.0' encoding='UTF-8'?><?xml-stylesheet
href="http://www.blogger.com/styles/atom.css"; type="text/css"?>
+<!--
+
+ 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.
+
+-->
+<feed xmlns='http://www.w3.org/2005/Atom'
xmlns:openSearch='http://a9.com/-/spec/opensearchrss/1.0/'>
+ <id>tag:blogger.com,1999:blog-637417304187784899</id>
+ <updated>2007-12-01T12:33:31.529Z</updated>
+ <title type='text'>James Strachan's Blog</title>
+ <link rel='alternate' type='text/html' href='http://macstrac.blogspot.com/'/>
+ <link rel='next' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default?start-index=26&max-results=25'/>
+ <link rel='http://schemas.google.com/g/2005#feed' type='application/atom+xml'
+ href='http://macstrac.blogspot.com/feeds/posts/default'/>
+ <link rel='self' type='application/atom+xml'
href='http://macstrac.blogspot.com/feeds/posts/default'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ <generator version='7.00' uri='http://www.blogger.com'>Blogger</generator>
+ <openSearch:totalResults>54</openSearch:totalResults>
+ <openSearch:startIndex>1</openSearch:startIndex>
+ <openSearch:itemsPerPage>25</openSearch:itemsPerPage>
+
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-597595982330434286</id>
+ <published>2007-11-26T08:46:00.000Z</published>
+ <updated>2025-11-26T08:49:10.354Z</updated>
+ <title type='text'>ActiveMQ webinar archive available</title>
+ <content type='html'>Our <a
href="http://macstrac.blogspot.com/2007/11/webcast-today-on-apache-activemq.html">previously
+ announced webinar on ActiveMQ</a> is now available on <a
+ href="http://open.iona.com/resources/video-archived-webinars/">IONA's
webinar and screencast archive</a>.
+ You can <a href="https://cc.readytalk.com/play?id=go5pfj81">view
it here</a>.
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/activemq-webinar-archive-available.html'
+ title='ActiveMQ webinar archive available'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=597595982330434286'
+ title='0 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/597595982330434286/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/597595982330434286'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/597595982330434286'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-6948835939296895803</id>
+ <published>2007-11-14T14:18:00.001Z</published>
+ <updated>2025-11-14T19:43:38.291Z</updated>
+ <title type='text'>More thoughts on RESTful Message Queues</title>
+ <content type='html'>Just a small follow up on my previous <span
style="font-size:100%;"><a
+ style="font-weight: normal;"
href="http://macstrac.blogspot.com/2007/08/pure-restful-api-to-activemq-via.html">Pure
+ RESTful API to ActiveMQ via AtomPub</a></span>. AtomPub
rocks and all - I was thinking whats the
+ easiest possible RESTful client to subscribe to a message queue.<br
/><blockquote></blockquote><blockquote>GET
+
queues/foo.bar?user=myUniqueId&amp;timeout=5000</blockquote>This
would activate my subscription to the
+ queue <span style="font-style: italic;">foo.bar</span> (or
help to keep it alive) and return my own
+ list of messages as an Atom feed that I am allowed to view and DELETE
when I have consumed them.<br /><br
+ />This operation is idempotent and would work great with proxies and
caches (assuming the right HTTP headers /
+ ETags stuff) letting clients to keep GETing as often as they like.
Though if you don't use the subscription for
+ period of time, your subscription can go stale and timeout; any messages
in your message collection could be
+ removed.<br /><br />The slightly smelly thing here is we
need to either use cookies (such as for HTTP
+ session handling with servlets) or use a custom session ID header in the
URL to uniquely differentiate the
+ subscriptions. Anyone got a better idea?<br /><br />I guess
we could demand that clients PUT/POST to
+ get a new Location URL on which to GET their subscriptions; but this
would require a custom REST client.
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/more-thoughts-on-restful-message-queues.html'
+ title='More thoughts on RESTful Message Queues'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=6948835939296895803'
+ title='3 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/6948835939296895803/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/6948835939296895803'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/6948835939296895803'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-8132857663967818867</id>
+ <published>2007-11-13T15:30:00.000Z</published>
+ <updated>2025-11-13T16:40:47.166Z</updated>
+ <category scheme='http://www.blogger.com/atom/ns#' term='camel'/>
+ <title type='text'>Feedback on my Camel talk at the IJTC conference</title>
+ <content type='html'>Thanks to <a
href="http://capedcrusading.blogspot.com/">John</a> for the <a
+
href="http://capedcrusading.blogspot.com/2007/11/ijtc-conference-spring-grails-camel.html">great
feedback on my
+ Camel talk</a> (and other talks too). I started to write a huge
reply and figured I'd post it here instead
+ then link to it as its easier to reply to different parts.<br
/><blockquote>Camel itself is not an ESB
+ per-se, it is a component of an ESB </blockquote>Agreed. The idea
is Camel does the routing and <a
+
href="http://activemq.apache.org/camel/enterprise-integration-patterns.html";
title="Enterprise Integration
+ Patterns">Enterprise Integration Patterns</a> - you can then
use it inside a web service stack like <span
+ class="nobr"></span><a
href="http://incubator.apache.org/cxf/"; title="Visit page outside Confluence"
+ rel="nofollow">Apache CXF</a>, a message broker like <span
style="text-decoration: underline;"></span><span
+ class="nobr"><a href="http://activemq.apache.org/"; title="Visit
page outside Confluence" rel="nofollow">Apache
+ ActiveMQ</a></span> or an ESB like<span
class="nobr"><a
+ href="http://incubator.apache.org/servicemix/"; title="Visit page outside
Confluence" rel="nofollow"> Apache
+ ServiceMix</a></span><br /><blockquote>1. One of
the key messages that came out of the
+ session was how Java-centric the Camel solution is - Strachan went so
far as to articulate the view that coding in
+ XML was a fundamentally bad idea.<br /></blockquote>Maybe
being the author of <a
+ href="http://commons.apache.org/jelly/">Jelly</a> and hating
writing XSLT has made me a bit too sensitive
+ to the idea of programming by XML. Quite a few customers I talk with
report frustration of too much XML hacking
+ with Spring (which is being addressed in Guice and Spring 2.5). But heck
if you like programming in XML be my
+ guest :)<br /><blockquote>He went on to cast aspersions on
graphical tools also. His clear preference
+ was that integration logic should be written in Java. Given that I have
worked with highly skilled Java developers
+ for many years now I was not too surprised to hear this - many good
developers shy away from tooling, seeing it as
+ compromising their style or the power of the underlying framework -
hence the longevity of vi and emacs I suppose.<br
+ /></blockquote>I didn't mean to cast aspersions on graphical
tools; more that pretty much all developers
+ understand Java these days, its pretty universal - whereas most complex
graphical tools require a fair amount of
+ learning to get used to them.<br /><br />Personally I
totally prefer writing in Java rather than XML
+ or using visual tooling but one of the main requirements of Camel is
that you can configure and specify routes in
+ any way you like - via a graphical IDE (e.g. Cimero) or via XML or Java
or Groovy or Ruby or one day hopefully a
+ real DSL. Using XML is quite useful as you can just drop your routing
rules inside a spring XML file such as
+ inside the <span style="text-decoration:
underline;"></span><span class="nobr"><a
+ href="http://activemq.apache.org/"; title="Visit page outside Confluence"
rel="nofollow">Apache ActiveMQ</a></span>.<br
+ /><br />So we're working hard to allow folks to specify routing
rules however they like - despite what I
+ prefer :). Irrespective of how you write your routing rules, we can <a
+ href="http://activemq.apache.org/camel/visualisation.html">visualise
them</a> so anyone can easily
+ understand them.<br /><blockquote>However it is important to
note that all developers are not
+ middleware experts and have no wish to be. Indeed the enterprises they
work for want their developers to spend as
+ little time and effort as possible on middleware plumbing. They need and
demand tools which will enable them to
+ get at least 80% of patterns done without having to understand the
middleware architecture, it's threading model,<br
+ /></blockquote>Thats one of the main things we're trying to
attempt with Camel - letting folks who are
+ not middleware experts easily use the <a
+
href="http://activemq.apache.org/camel/enterprise-integration-patterns.html";
title="Enterprise Integration
+ Patterns">Enterprise Integration Patterns</a> using a single
line of Java code - using <a
+ href="http://activemq.apache.org/camel/components.html">any transport
or component </a>with minimal
+ configuration required.<br /><blockquote>it's support for
configurable expression languages etc.<br
+ /></blockquote>I find it hard to understand how any tool can be
usable by folks without some kind of
+ expression language that they understand - whether its Java or SQL or
XPath/XQuery or whatnot. Even a visual query
+ definer is a language that users need to know.<br
/><blockquote>Most people express their requirements
+ in a declarative way- I have data at A that needs to get to B, on it's
way I need to perform transformation,
+ validation, logging etc. (indeed the EIP book itself does this) .
However Camel has taken a very Java centric
+ approach and I think this increases it's complexity unnecessarily.<br
/></blockquote>We are actually
+ trying to be as declarative as possible - don't let the fact that you
can use Java as the DSL confuse you. e.g.
+ here's what you just described in Camel using a single line of Java
code...<br /><blockquote>from(a).<br
+ />to("xsd:org/cheese/something.xsd").<br
/>to("xslt:com/acme/mytransform.xsl").<br
/>to("log:com.acme.MyLogger").<br
+ />to(b);<br /></blockquote>Its hard to be more concise
than that in Java code. But sure - you could
+ use some other language or XML or UI tool etc<br
/><blockquote>2. Camel presents transports as
+ nice-simple looking endpoint URIs in Java. However configuration of
these transports may not be as simple as it
+ seems. There appears to me to be a potential disconnection between the
Camel processors and the Camel Components
+ in terms of configuration.<br /></blockquote>You can always
configure anything in Camel via Java or
+ Spring; a component, endpoint, processor etc. The URI is just a
shorthand notation for configurating things; which
+ tends to work well with endpoints as usually all the smart configuration
is in the component.<br /><blockquote>In
+ the presentation processors use endpoints which can be configured using
Spring within Camel. However it would
+ appear that many of the component implementations are inherited from
ServiceMix and these properties will need to
+ be set presumably within the ServiceMix container configuration?
</blockquote>Not so - all the components
+ are configured in Spring via Camel. If you want to talk to ServiceMix
components and endpoints you can use the NMR
+ and the <a href="http://activemq.apache.org/camel/jbi.html">JBI
endpoint</a>.<br /><blockquote>When
+ you have multiple XML configurations to use a transport then the pretty
looking URI is hiding a lot of complexity
+ under the hood.<br /></blockquote>I hear you. We've tried
very hard to make things as easy as is
+ possible with minimal configuration; for example most configuration
tends to be on the Component rather than the
+ Endpoint. But you can configure things however you like in Java code or
Guice or Spring.<br /><blockquote>3.
+ Some of the pattern implementations look a little short on credibility -
take the aggregator pattern for example
+ the "aggregator" pattern does not seem to have the concept of a store -
so in essence one must have access to all
+ of the messages which require aggregation or must hold aggregations in
memory for the configurable timeout period.
+ This will clearly not work in a scalable way. Likewise as regards
clustering - if you are looking to aggregate 2
+ messages and they turn up on different servers you gotta
problem.</blockquote>You're right I purposely
+ missed out some of the detail on a few slides (such as specifying some
kind of persistence store or strategy for
+ aggregator or for idempotent consumer) but that is easily done via a
pluggable strategies Spring beans. However
+ just because I missed out some detail on some slides (its kinda hard in
an hour to present all the detail in all
+ the patterns as well as the rest of Camel) please don't think that
somehow Camel isn't short on capabilities.<br
+ /><blockquote>To summarise Camel is clearly a worthy set of
widgets and they will work for very simple
+ applications without Enterprise requirements.<br
/></blockquote>Ouch :). We've actually lots of
+ customers using Camel today in production with very Enterprise
requirements. <span class="nobr"></span><a
+ href="http://activemq.apache.org/"; title="Visit page outside Confluence"
rel="nofollow">Apache ActiveMQ</a>
+ 5.0 actually ships <a
href="http://activemq.apache.org/enterprise-integration-patterns.html">fully
+ integrated with Camel</a> so we've tons of users in production
using Camel today.<br /><br />I
+ hope I've managed to straighten out some misunderstandings on Camel; its
definitely a great fit for enterprise
+ requirements. Thanks for your great feedback - I'll definitely take it
on board on future presentations and try
+ and avoid confusing other folks :).
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/feedback-on-my-camel-talk-at-ijtc.html'
+ title='Feedback on my Camel talk at the IJTC conference'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=8132857663967818867'
+ title='2 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/8132857663967818867/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/8132857663967818867'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/8132857663967818867'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-2980017428692756666</id>
+ <published>2007-11-13T13:33:00.000Z</published>
+ <updated>2007-11-13T13:35:25.014Z</updated>
+ <title type='text'>webcast today on Apache ActiveMQ</title>
+ <content type='html'>Sorry for the really late notice - bad James! - but
I<a
+ href="http://open.iona.com/resources/news/#webinar">'m doing a
webinar today</a> on <a
+ href="http://activemq.apache.org/">Apache ActiveMQ</a> with my
fellow committer <a
+ href="http://hiramchirino.com/index.html">Hiram Chirino</a>.
Feel free to pop by and join us - or if
+ you're snowed you can catch the recording later on.
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/webcast-today-on-apache-activemq.html'
+ title='webcast today on Apache ActiveMQ'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=2980017428692756666'
+ title='0 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/2980017428692756666/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/2980017428692756666'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/2980017428692756666'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-8437027470203977326</id>
+ <published>2007-11-13T13:10:00.000Z</published>
+ <updated>2007-11-13T13:32:23.119Z</updated>
+ <title type='text'>my slides on ActiveMQ and Camel from last weeks Dublin
Conference</title>
+ <content type='html'>They are not that useful if you missed me talk, as
they are low on bullet points and high on
+ pictures :) But if you were there, here are the slides in PDF
format...<br /><ul><li><pre><a
+
href="http://people.apache.org/%7Ejstrachan/talks/ActiveMQ-Dublin07.pdf">ActiveMQ-Dublin07.pdf</a></pre></li><li><pre><a
+
href="http://people.apache.org/%7Ejstrachan/talks/Camel-Dublin07.pdf">Camel-Dublin07.pdf</a>
<br /></pre></li></ul></content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/my-slides-on-activemq-and-camel-from.html'
+ title='my slides on ActiveMQ and Camel from last weeks Dublin
Conference'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=8437027470203977326'
+ title='0 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/8437027470203977326/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/8437027470203977326'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/8437027470203977326'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-4574479486073739581</id>
+ <published>2007-11-13T13:09:00.000Z</published>
+ <updated>2007-11-13T13:10:23.292Z</updated>
+ <title type='text'>a great presentation on REST, JAX-WS and JSR 311</title>
+ <content type='html'>A great <a
+
href="http://www.innoq.com/blog/st/presentations/2007/2007-11-06-JSR-311-W-JAX.pdf";
class="jive-link-external">presentation</a>
+ from <a href="http://www.innoq.com/blog/st/";
class="jive-link-external">Stefan Tilkov</a> on REST,
+ JAX-WS and JSR 311; highly recommended.
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/great-presentation-on-rest-jax-ws-and.html'
+ title='a great presentation on REST, JAX-WS and JSR 311'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=4574479486073739581'
+ title='0 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/4574479486073739581/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/4574479486073739581'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/4574479486073739581'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+ <entry>
+
<id>tag:blogger.com,1999:blog-637417304187784899.post-459928383168678386</id>
+ <published>2007-11-05T10:34:00.000Z</published>
+ <updated>2007-11-05T10:46:07.867Z</updated>
+ <title type='text'>Speaking at the Irish Java Technology Conference on
Thursday and Friday</title>
+ <content type='html'>I'll be <a
href="http://ijtc.firstport.ie/bio.aspx?sid=33">speaking</a> this week
+ at the <a href="http://ijtc.firstport.ie/"><span
id="cmsContent"></span>Irish Java Technology
+ Conference</a> this week on Thursday and Friday. My talks
are<br /><ul><li><a
+ href="http://ijtc.firstport.ie/lecture.aspx?lid=34">Enterprise
Messaging With Apache ActiveMQ</a></li><li><a
+ href="http://ijtc.firstport.ie/lecture.aspx?lid=14">Easy Enterprise
Integration Patterns with Apache Camel,
+ ActiveMQ and ServiceMix</a></li></ul>Do pop along and
say hi if you're gonna be in the Dublin
+ area this week.
+ </content>
+ <link rel='alternate' type='text/html'
+
href='http://macstrac.blogspot.com/2007/11/speaking-at-irish-java-technology.html'
+ title='Speaking at the Irish Java Technology Conference on Thursday
and Friday'/>
+ <link rel='replies' type='text/html'
+
href='http://www.blogger.com/comment.g?blogID=637417304187784899&postID=459928383168678386'
+ title='3 Comments'/>
+ <link rel='replies' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/459928383168678386/comments/default'
title='Post Comments'/>
+ <link rel='self' type='application/atom+xml'
+
href='http://macstrac.blogspot.com/feeds/posts/default/459928383168678386'/>
+ <link rel='edit' type='application/atom+xml'
+
href='http://www.blogger.com/feeds/637417304187784899/posts/default/459928383168678386'/>
+ <author>
+ <name>James Strachan</name>
+ </author>
+ </entry>
+</feed>
\ No newline at end of file
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogDefaultIdempotencyTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogDefaultIdempotencyTest.java
new file mode 100644
index 000000000000..e1bcb7e34cd2
--- /dev/null
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogDefaultIdempotencyTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.component.atom;
+
+import org.apache.camel.*;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Atom Good Blog default idempotency test
+ */
+@DisabledOnOs(OS.AIX)
+public class AtomGoodBlogDefaultIdempotencyTest {
+
+ // START SNIPPET: e1
+
+ // This is the CamelContext that is the heart of Camel
+ private CamelContext context;
+ private static final String ROUTE_ID = "1234";
+
+ protected CamelContext createCamelContext() throws Exception {
+ // Create CamelContext and register test routes using the route
builder DSL syntax
+ context = new DefaultCamelContext();
+ context.addRoutes(createMyRoutes());
+ return context;
+ }
+
+ /**
+ * This is the route builder where we create our routes using the Camel DSL
+ */
+ protected RouteBuilder createMyRoutes() {
+ return new RouteBuilder() {
+ public void configure() {
+ // Idempotent using default strategy in combination of small
delay and throttleEntries=false, ensures that we
+ // only process each entry once, even though the full blog is
read multiple times
+
from("atom:file:src/test/data/feed.atom?splitEntries=true&throttleEntries=false&delay=1000").routeId(ROUTE_ID)
+ .to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * This test verifies that the correct idempotent strategy is used
+ */
+ @Test
+ void testDefaultIdempotentStrategy() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+
+ AtomEndpoint endpoint = context.getEndpoint("atom:*",
AtomEndpoint.class);
+ assertTrue(endpoint.isIdempotent());
+ assertInstanceOf(ItemUpdatedIdempotentStrategy.class,
endpoint.getIdempotentStrategy());
+
+ // stop Camel after use
+ context.stop();
+ }
+
+ @Test
+ void testIdempotentFilterFullRead() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ context.start();
+
+ // Get the mock endpoint
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+ Thread.sleep(5000);
+
+ // There should be at least two good blog entries from the feed
+ mock.expectedMessageCount(7);
+
+ // Assert
+ mock.assertIsSatisfied();
+
+ // stop Camel after use
+ context.stop();
+ }
+
+ @Test
+ void testIdempotentFilterSimulateFeedUpdate() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ context.start();
+
+ // Get the mock endpoint
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+
+ // Read all 7 items from the feed...
+ mock.expectedMessageCount(7);
+ mock.assertIsSatisfied();
+
+ // Simulate feed update by resetting the file...
+ AtomEntryPollingConsumer atomConsumer = (AtomEntryPollingConsumer)
context.getRoute(ROUTE_ID).getConsumer();
+
atomConsumer.getEndpoint().setFeedUri("file:src/test/data/updated-feed.atom");
+
+ // Read the three updated entries
+ mock.expectedMessageCount(10);
+ mock.assertIsSatisfied();
+
+ // stop Camel after use
+ context.stop();
+ }
+ // END SNIPPET: e1
+
+}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogNotIdempotentTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogNotIdempotentTest.java
new file mode 100644
index 000000000000..891fd52d090b
--- /dev/null
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogNotIdempotentTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.component.atom;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+/**
+ * Atom Good Blog non-idempotent test
+ */
+@DisabledOnOs(OS.AIX)
+public class AtomGoodBlogNotIdempotentTest {
+
+ // This is the CamelContext that is the heart of Camel
+ private CamelContext context;
+ private static final String ROUTE_ID = "1234";
+
+ protected CamelContext createCamelContext() throws Exception {
+ // Create CamelContext and register test routes using the route
builder DSL syntax
+ context = new DefaultCamelContext();
+ context.addRoutes(createMyRoutes());
+ return context;
+ }
+
+ /**
+ * This is the route builder where we create our routes using the Camel DSL
+ */
+ protected RouteBuilder createMyRoutes() {
+ return new RouteBuilder() {
+ public void configure() {
+ // Atom read setup as not idempotent
+
from("atom:file:src/test/data/feed.atom?splitEntries=true&throttleEntries=false&delay=1000&idempotent=false")
+ .routeId(ROUTE_ID)
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void testIdempotentFilterFullRead() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ context.start();
+
+ // Get the mock endpoint
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+
+ // The blot should be polled least two times
+ mock.expectedMinimumMessageCount(14);
+
+ // Assert
+ mock.assertIsSatisfied();
+
+ // stop Camel after use
+ context.stop();
+ }
+}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsCustomItempotencyTest.java
similarity index 57%
copy from
components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
copy to
components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsCustomItempotencyTest.java
index a8267cc77f5d..d79874e48d67 100644
---
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsCustomItempotencyTest.java
@@ -18,7 +18,6 @@ package org.apache.camel.component.atom;
import com.apptasticsoftware.rssreader.Item;
import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.impl.DefaultCamelContext;
@@ -27,13 +26,13 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
/**
- * Example for wiki documentation
+ * Atom Good Blog custom idempotency test
*/
@DisabledOnOs(OS.AIX)
-public class AtomGoodBlogsTest {
-
- // START SNIPPET: e1
+public class AtomGoodBlogsCustomItempotencyTest {
// This is the CamelContext that is the heart of Camel
private CamelContext context;
@@ -42,7 +41,7 @@ public class AtomGoodBlogsTest {
// First we register a blog service in our bean registry
SimpleRegistry registry = new SimpleRegistry();
- registry.bind("blogService", new BlogService());
+ registry.bind("myIdempotentStrategy", new CustomIdempotentStrategy());
// Then we create the camel context with our bean registry
context = new DefaultCamelContext(registry);
@@ -59,28 +58,25 @@ public class AtomGoodBlogsTest {
protected RouteBuilder createMyRoutes() {
return new RouteBuilder() {
public void configure() {
- // We pool the atom feeds from the source for further
processing in the seda queue
- // we set the delay to 1 second for each pool as this is a
unit test also, and we can
- // not wait the default poll interval of 60 seconds.
- // Using splitEntries=true will during polling only fetch one
Atom Entry at any given time.
- // As the feed.atom file contains 7 entries, using this will
require 7 polls to fetch the entire
- // content. When Camel have reach the end of entries it will
refresh the atom feed from URI source
- // and restart - but as Camel by default uses the
UpdatedDateFilter it will only deliver new
- // blog entries to "seda:feeds". So only when James Straham
updates his blog with a new entry
- // Camel will create an exchange for the seda:feeds.
-
from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=1000").to("seda:feeds");
-
- // From the feeds we filter each blot entry by using our blog
service class
- from("seda:feeds").filter().method("blogService",
"isGoodBlog").to("seda:goodBlogs");
-
- // And the good blogs is moved to a mock queue as this sample
is also used for unit testing
- // this is one of the strengths in Camel that you can also use
the mock endpoint for your
- // unit tests
- from("seda:goodBlogs").to("mock:result");
+ // Combination of large delay and throttleEntries=false
ensures that we only process the full feed once
+
from("atom:file:src/test/data/feed.atom?delay=10000&splitEntries=true&throttleEntries=false&idempotentStrategy=#myIdempotentStrategy")
+ .to("mock:result");
}
};
}
+ /**
+ * This test verifies that the correct idempotent strategy is used
+ */
+ @Test
+ void testCustomIdempotentStrategy() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ AtomEndpoint endpoint
+ =
context.getEndpoint("atom:feed.atom?idempotentStrategy=#myIdempotentStrategy",
AtomEndpoint.class);
+ assertInstanceOf(CustomIdempotentStrategy.class,
endpoint.getIdempotentStrategy());
+ }
+
/**
* This is the actual junit test method that does the assertion that our
routes is working as expected
*/
@@ -94,7 +90,7 @@ public class AtomGoodBlogsTest {
MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
// There should be at least two good blog entries from the feed
- mock.expectedMinimumMessageCount(2);
+ mock.expectedMessageCount(2);
// Asserts that the above expectations is true, will throw assertions
exception if it failed
// Camel will default wait max 20 seconds for the assertions to be
true, if the conditions
@@ -104,25 +100,13 @@ public class AtomGoodBlogsTest {
// stop Camel after use
context.stop();
}
+}
- /**
- * Services for blogs
- */
- public static class BlogService {
-
- /**
- * Tests the blogs if its a good blog entry or not
- */
- public boolean isGoodBlog(Exchange exchange) {
- Item entry = exchange.getIn().getBody(Item.class);
- String title = entry.getTitle().get();
-
- // We like blogs about Camel
- return title.toLowerCase().contains("camel");
- }
+class CustomIdempotentStrategy implements AtomIdempotentStrategy {
+ @Override
+ public boolean isValidItem(Item entry) {
+ String ONLY_INTERESTING_TOPIC = "ActiveMQ";
+ return entry.getTitle().orElse("").contains(ONLY_INTERESTING_TOPIC);
}
-
- // END SNIPPET: e1
-
}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyCustomRepositoryTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyCustomRepositoryTest.java
new file mode 100644
index 000000000000..af34c9b695f7
--- /dev/null
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyCustomRepositoryTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.component.atom;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.SimpleRegistry;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+/**
+ * Atom Good Blog repository idempotency test
+ */
+@DisabledOnOs(OS.AIX)
+public class AtomGoodBlogsRepositoryItempotencyCustomRepositoryTest {
+
+ // This is the CamelContext that is the heart of Camel
+ private CamelContext context;
+
+ protected CamelContext createCamelContext() throws Exception {
+ // First we register a IdempotentRepository simulating several reads
from the same blog
+ SimpleRegistry registry = new SimpleRegistry();
+ IdempotentRepository anyCustomrepository =
MemoryIdempotentRepository.memoryIdempotentRepository(100);
+
anyCustomrepository.add("tag:blogger.com,1999:blog-637417304187784899.post-459928383168678386");
+
anyCustomrepository.add("tag:blogger.com,1999:blog-637417304187784899.post-4574479486073739581");
+
anyCustomrepository.add("tag:blogger.com,1999:blog-637417304187784899.post-8437027470203977326");
+
anyCustomrepository.add("tag:blogger.com,1999:blog-637417304187784899.post-2980017428692756666");
+ registry.bind("myIdempotentRepository", anyCustomrepository);
+
+ // Then we create the camel context with our bean registry
+ context = new DefaultCamelContext(registry);
+
+ // Then we add all the routes we need using the route builder DSL
syntax
+ context.addRoutes(createMyRoutes());
+
+ return context;
+ }
+
+ /**
+ * This is the route builder where we create our routes using the Camel DSL
+ */
+ protected RouteBuilder createMyRoutes() {
+ return new RouteBuilder() {
+ public void configure() {
+ // Idempotent using repository strategy in combination of
small delay and throttleEntries=false, ensures that we
+ // only process each entry once, even though the full blog is
read multiple times
+
from("atom:file:src/test/data/feed.atom?splitEntries=true&throttleEntries=false&delay=1000&idempotentStrategy=repository&idempotentRepository=#myIdempotentRepository")
+ .to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * This test verifies that the correct idempotent strategy is used
+ */
+ @Test
+ void testDefaultIdempotentStrategy() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+
+ AtomEndpoint endpoint =
context.getEndpoint("atom:feed.atom?idempotentStrategy=repository",
AtomEndpoint.class);
+ assertInstanceOf(RepositoryGuidIdempotentStrategy.class,
endpoint.getIdempotentStrategy());
+
+ // stop Camel after use
+ context.stop();
+ }
+
+ /**
+ * This is the actual junit test method that does the assertion that our
routes is working as expected
+ */
+ @Test
+ void testFiltering() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ context.start();
+
+ // Get the mock endpoint
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+
+ // There should be at only three good blog entries, not already in
customRepository
+ mock.expectedMessageCount(3);
+
+ // Make sure that the component has time for several reads
+ Thread.sleep(3000);
+
+ // Assert
+ mock.assertIsSatisfied(3000);
+
+ // stop Camel after use
+ context.stop();
+ }
+}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyTest.java
similarity index 52%
copy from
components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
copy to
components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyTest.java
index a8267cc77f5d..78b66e44fe25 100644
---
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsRepositoryItempotencyTest.java
@@ -16,9 +16,7 @@
*/
package org.apache.camel.component.atom;
-import com.apptasticsoftware.rssreader.Item;
import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.impl.DefaultCamelContext;
@@ -27,13 +25,13 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
/**
- * Example for wiki documentation
+ * Atom Good Blog repository idempotency test
*/
@DisabledOnOs(OS.AIX)
-public class AtomGoodBlogsTest {
-
- // START SNIPPET: e1
+public class AtomGoodBlogsRepositoryItempotencyTest {
// This is the CamelContext that is the heart of Camel
private CamelContext context;
@@ -42,7 +40,6 @@ public class AtomGoodBlogsTest {
// First we register a blog service in our bean registry
SimpleRegistry registry = new SimpleRegistry();
- registry.bind("blogService", new BlogService());
// Then we create the camel context with our bean registry
context = new DefaultCamelContext(registry);
@@ -59,70 +56,51 @@ public class AtomGoodBlogsTest {
protected RouteBuilder createMyRoutes() {
return new RouteBuilder() {
public void configure() {
- // We pool the atom feeds from the source for further
processing in the seda queue
- // we set the delay to 1 second for each pool as this is a
unit test also, and we can
- // not wait the default poll interval of 60 seconds.
- // Using splitEntries=true will during polling only fetch one
Atom Entry at any given time.
- // As the feed.atom file contains 7 entries, using this will
require 7 polls to fetch the entire
- // content. When Camel have reach the end of entries it will
refresh the atom feed from URI source
- // and restart - but as Camel by default uses the
UpdatedDateFilter it will only deliver new
- // blog entries to "seda:feeds". So only when James Straham
updates his blog with a new entry
- // Camel will create an exchange for the seda:feeds.
-
from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=1000").to("seda:feeds");
-
- // From the feeds we filter each blot entry by using our blog
service class
- from("seda:feeds").filter().method("blogService",
"isGoodBlog").to("seda:goodBlogs");
-
- // And the good blogs is moved to a mock queue as this sample
is also used for unit testing
- // this is one of the strengths in Camel that you can also use
the mock endpoint for your
- // unit tests
- from("seda:goodBlogs").to("mock:result");
+ // Idempotent using repository strategy in combination of
small delay and throttleEntries=false, ensures that we
+ // only process each entry once, even though the full blog is
read multiple times
+
from("atom:file:src/test/data/feed.atom?splitEntries=true&throttleEntries=false&delay=1000&idempotentStrategy=repository")
+ .to("mock:result");
}
};
}
/**
- * This is the actual junit test method that does the assertion that our
routes is working as expected
+ * This test verifies that the correct idempotent strategy is used
*/
@Test
- void testFiltering() throws Exception {
+ void testDefaultIdempotentStrategy() throws Exception {
// create and start Camel
context = createCamelContext();
- context.start();
-
- // Get the mock endpoint
- MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
-
- // There should be at least two good blog entries from the feed
- mock.expectedMinimumMessageCount(2);
- // Asserts that the above expectations is true, will throw assertions
exception if it failed
- // Camel will default wait max 20 seconds for the assertions to be
true, if the conditions
- // is true sooner Camel will continue
- mock.assertIsSatisfied();
+ AtomEndpoint endpoint =
context.getEndpoint("atom:feed.atom?idempotentStrategy=repository",
AtomEndpoint.class);
+ assertInstanceOf(RepositoryGuidIdempotentStrategy.class,
endpoint.getIdempotentStrategy());
// stop Camel after use
context.stop();
}
/**
- * Services for blogs
+ * This is the actual junit test method that does the assertion that our
routes is working as expected
*/
- public static class BlogService {
+ @Test
+ void testFiltering() throws Exception {
+ // create and start Camel
+ context = createCamelContext();
+ context.start();
- /**
- * Tests the blogs if its a good blog entry or not
- */
- public boolean isGoodBlog(Exchange exchange) {
- Item entry = exchange.getIn().getBody(Item.class);
- String title = entry.getTitle().get();
+ // Get the mock endpoint
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
- // We like blogs about Camel
- return title.toLowerCase().contains("camel");
- }
+ // There should be at least two good blog entries from the feed
+ mock.expectedMessageCount(7);
- }
+ // Make sure that the component has time for several reads
+ Thread.sleep(3000);
- // END SNIPPET: e1
+ // Assert
+ mock.assertIsSatisfied(3000);
+ // stop Camel after use
+ context.stop();
+ }
}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
index a8267cc77f5d..f90782461d50 100644
---
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomGoodBlogsTest.java
@@ -65,7 +65,7 @@ public class AtomGoodBlogsTest {
// Using splitEntries=true will during polling only fetch one
Atom Entry at any given time.
// As the feed.atom file contains 7 entries, using this will
require 7 polls to fetch the entire
// content. When Camel have reach the end of entries it will
refresh the atom feed from URI source
- // and restart - but as Camel by default uses the
UpdatedDateFilter it will only deliver new
+ // and restart - but as the Camel Atom component is idempotent
by default, it will only deliver new
// blog entries to "seda:feeds". So only when James Straham
updates his blog with a new entry
// Camel will create an exchange for the seda:feeds.
from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=1000").to("seda:feeds");
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategyTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategyTest.java
new file mode 100644
index 000000000000..0f480040005f
--- /dev/null
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/ItemUpdatedIdempotentStrategyTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.component.atom;
+
+import java.text.ParseException;
+import java.util.Date;
+import java.util.Optional;
+
+import com.apptasticsoftware.rssreader.Item;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+public class ItemUpdatedIdempotentStrategyTest {
+
+ /**
+ * This class tests the isValidItem method in the
ItemUpdatedIdempotentStrategy class. The method determines whether
+ * an RSS item is valid based on its updated or published date in relation
to the previously processed item.
+ */
+
+ @Test
+ void testValidEntryWithNewerUpdatedDate() throws ParseException {
+ // Arrange
+ ItemUpdatedIdempotentStrategy strategy = new
ItemUpdatedIdempotentStrategy();
+ Item item = Mockito.mock(Item.class);
+
+ Date olderUpdate = AtomConverter.toDate("2023-01-01T00:00:00Z");
+ Date newerUpdate = AtomConverter.toDate("2023-02-01T00:00:00Z");
+
when(item.getUpdated()).thenReturn(Optional.of("2023-02-01T00:00:00Z"));
+
+ // Act and Assert
+ assertTrue(strategy.isValidItem(item));
+
when(item.getUpdated()).thenReturn(Optional.of("2023-01-01T00:00:00Z"));
+ assertFalse(strategy.isValidItem(item));
+ }
+
+ @Test
+ void testValidEntryWithPubDateFallback() throws ParseException {
+ // Arrange
+ ItemUpdatedIdempotentStrategy strategy = new
ItemUpdatedIdempotentStrategy();
+ Item item = Mockito.mock(Item.class);
+
+ when(item.getUpdated()).thenReturn(Optional.empty());
+
when(item.getPubDate()).thenReturn(Optional.of("2023-03-01T00:00:00Z"));
+
+ // Act and Assert
+ assertTrue(strategy.isValidItem(item));
+
+
when(item.getPubDate()).thenReturn(Optional.of("2023-01-01T00:00:00Z"));
+ assertFalse(strategy.isValidItem(item));
+ }
+
+ @Test
+ void testValidEntryWithNoDatesAvailable() {
+ // Arrange
+ Item item = Mockito.mock(Item.class);
+ when(item.getUpdated()).thenReturn(Optional.empty());
+ when(item.getPubDate()).thenReturn(Optional.empty());
+
+ ItemUpdatedIdempotentStrategy strategy = new
ItemUpdatedIdempotentStrategy();
+
+ // Act and Assert
+ assertTrue(strategy.isValidItem(item));
+ }
+
+ @Test
+ void testEntryWithInvalidDateString() {
+ // Arrange
+ Item item = Mockito.mock(Item.class);
+ when(item.getUpdated()).thenReturn(Optional.of("invalid-date"));
+ when(item.getPubDate()).thenReturn(Optional.of("invalid-date"));
+
+ ItemUpdatedIdempotentStrategy strategy = new
ItemUpdatedIdempotentStrategy();
+
+ // Act
+ boolean isValid = strategy.isValidItem(item);
+
+ // Assert
+ assertTrue(isValid);
+ }
+
+ @Test
+ void testEntryWithEqualDate() throws ParseException {
+ // Arrange
+ Item item1 = Mockito.mock(Item.class);
+ Item item2 = Mockito.mock(Item.class);
+ ItemUpdatedIdempotentStrategy strategy = new
ItemUpdatedIdempotentStrategy();
+
+ Date date = AtomConverter.toDate("2023-01-01T00:00:00Z");
+
when(item1.getUpdated()).thenReturn(Optional.of("2023-01-01T00:00:00Z"));
+
when(item2.getUpdated()).thenReturn(Optional.of("2023-01-01T00:00:00Z"));
+
+ // Act and Assert
+ // First item is valid
+ assertTrue(strategy.isValidItem(item1));
+
+ // Second item is invalid because it has the same updated date
+ assertFalse(strategy.isValidItem(item2));
+ }
+}
diff --git
a/components/camel-atom/src/test/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategyTest.java
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategyTest.java
new file mode 100644
index 000000000000..eed334b958a4
--- /dev/null
+++
b/components/camel-atom/src/test/java/org/apache/camel/component/atom/RepositoryGuidIdempotentStrategyTest.java
@@ -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.
+ */
+package org.apache.camel.component.atom;
+
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import com.apptasticsoftware.rssreader.Item;
+import com.apptasticsoftware.rssreader.RssReader;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RepositoryGuidIdempotentStrategyTest {
+
+ @Test
+ void testIsValidItemReturnsTrueForNullEntry() {
+ RepositoryGuidIdempotentStrategy strategy = new
RepositoryGuidIdempotentStrategy();
+
+ assertTrue(strategy.isValidItem(null));
+ }
+
+ @Test
+ void testIsValidItemAcceptsFirstGuidAndRejectsDuplicate() throws Exception
{
+ RepositoryGuidIdempotentStrategy strategy = new
RepositoryGuidIdempotentStrategy();
+ Item item = loadItemsWithGuid(1).get(0);
+
+ assertTrue(strategy.isValidItem(item), "First time GUID should be
accepted");
+ assertFalse(strategy.isValidItem(item), "Duplicate GUID should be
rejected");
+ }
+
+ @Test
+ void testIsValidItemAcceptsDifferentGuids() throws Exception {
+ RepositoryGuidIdempotentStrategy strategy = new
RepositoryGuidIdempotentStrategy();
+ List<Item> items = loadItemsWithGuid(2);
+
+ assertTrue(strategy.isValidItem(items.get(0)), "First GUID should be
accepted");
+ assertTrue(strategy.isValidItem(items.get(1)), "Second, different GUID
should also be accepted");
+ }
+
+ private static List<Item> loadItemsWithGuid(int count) throws Exception {
+ try (InputStream is =
Files.newInputStream(Path.of("src/test/data/feed.atom"))) {
+ return new RssReader()
+ .read(is)
+ .filter(item -> item.getGuid().isPresent())
+ .limit(count)
+ .collect(Collectors.toList());
+ }
+ }
+}
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AtomEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AtomEndpointBuilderFactory.java
index 829f58b56602..5377305f2103 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AtomEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AtomEndpointBuilderFactory.java
@@ -174,6 +174,80 @@ public interface AtomEndpointBuilderFactory {
doSetProperty("throttleEntries", throttleEntries);
return this;
}
+ /**
+ * Option to use the Idempotent Consumer EIP pattern to let Camel skip
+ * already processed entries. By default it skips Atom entries that is
+ * older that the last read entry, using Atom updatedDate and
+ * publishedDate, but this strategy can be changed with the
+ * idempotentStrategy option. Idempotency only works when splitEntries
=
+ * true.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: filter
+ *
+ * @param idempotent the value to set
+ * @return the dsl builder
+ */
+ default AtomEndpointBuilder idempotent(boolean idempotent) {
+ doSetProperty("idempotent", idempotent);
+ return this;
+ }
+ /**
+ * Option to use the Idempotent Consumer EIP pattern to let Camel skip
+ * already processed entries. By default it skips Atom entries that is
+ * older that the last read entry, using Atom updatedDate and
+ * publishedDate, but this strategy can be changed with the
+ * idempotentStrategy option. Idempotency only works when splitEntries
=
+ * true.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: filter
+ *
+ * @param idempotent the value to set
+ * @return the dsl builder
+ */
+ default AtomEndpointBuilder idempotent(String idempotent) {
+ doSetProperty("idempotent", idempotent);
+ return this;
+ }
+ /**
+ * A pluggable repository org.apache.camel.spi.IdempotentRepository
+ * which by default use MemoryIdempotentRepository if none is specified
+ * and idempotent is true.
+ *
+ * The option is a:
+ * <code>org.apache.camel.spi.IdempotentRepository</code> type.
+ *
+ * Group: filter
+ *
+ * @param idempotentRepository the value to set
+ * @return the dsl builder
+ */
+ default AtomEndpointBuilder
idempotentRepository(org.apache.camel.spi.IdempotentRepository
idempotentRepository) {
+ doSetProperty("idempotentRepository", idempotentRepository);
+ return this;
+ }
+ /**
+ * A pluggable repository org.apache.camel.spi.IdempotentRepository
+ * which by default use MemoryIdempotentRepository if none is specified
+ * and idempotent is true.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.spi.IdempotentRepository</code> type.
+ *
+ * Group: filter
+ *
+ * @param idempotentRepository the value to set
+ * @return the dsl builder
+ */
+ default AtomEndpointBuilder idempotentRepository(String
idempotentRepository) {
+ doSetProperty("idempotentRepository", idempotentRepository);
+ return this;
+ }
/**
* The number of subsequent error polls (failed due some error) that
* should happen before the backoffMultipler should kick-in.
@@ -826,6 +900,56 @@ public interface AtomEndpointBuilderFactory {
doSetProperty("feedHeader", feedHeader);
return this;
}
+ /**
+ * A pluggable strategy
+ * org.apache.camel.component.FeedIdempotentStrategy to use when
+ * checking idempotency. Camel provides two implementations out of the
+ * box: default and repository. The updated strategy is used as default
+ * if idempotent = true. The default strategy checks the Atom entrys
+ * updated or published date is newer than the previously read entry.
+ * You can provide your own implementation of the
+ * org.apache.camel.component.FeedIdempotentStrategy and refer to it
+ * using the # notation.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.atom.AtomIdempotentStrategy</code>
+ * type.
+ *
+ * Default: default
+ * Group: filter (advanced)
+ *
+ * @param idempotentStrategy the value to set
+ * @return the dsl builder
+ */
+ default AdvancedAtomEndpointBuilder
idempotentStrategy(org.apache.camel.component.atom.AtomIdempotentStrategy
idempotentStrategy) {
+ doSetProperty("idempotentStrategy", idempotentStrategy);
+ return this;
+ }
+ /**
+ * A pluggable strategy
+ * org.apache.camel.component.FeedIdempotentStrategy to use when
+ * checking idempotency. Camel provides two implementations out of the
+ * box: default and repository. The updated strategy is used as default
+ * if idempotent = true. The default strategy checks the Atom entrys
+ * updated or published date is newer than the previously read entry.
+ * You can provide your own implementation of the
+ * org.apache.camel.component.FeedIdempotentStrategy and refer to it
+ * using the # notation.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.atom.AtomIdempotentStrategy</code>
+ * type.
+ *
+ * Default: default
+ * Group: filter (advanced)
+ *
+ * @param idempotentStrategy the value to set
+ * @return the dsl builder
+ */
+ default AdvancedAtomEndpointBuilder idempotentStrategy(String
idempotentStrategy) {
+ doSetProperty("idempotentStrategy", idempotentStrategy);
+ return this;
+ }
}
public interface AtomBuilders {
