You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2022/04/19 16:40:34 UTC
[camel] branch main updated: CAMEL-17963: added resume API support for camel-atom and camel-rss
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new c5218f1bdbf CAMEL-17963: added resume API support for camel-atom and camel-rss
c5218f1bdbf is described below
commit c5218f1bdbf1a59d0bb15b2d589a86436e8d4294
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Tue Apr 19 12:31:16 2022 +0200
CAMEL-17963: added resume API support for camel-atom and camel-rss
---
.../component/atom/AtomEndpointConfigurer.java | 9 -----
.../component/atom/AtomEndpointUriFactory.java | 4 +-
.../org/apache/camel/component/atom/atom.json | 2 -
.../apache/camel/component/atom/AtomEndpoint.java | 6 +--
.../component/atom/AtomEntryPollingConsumer.java | 12 +-----
.../camel/component/atom/UpdatedDateFilter.java | 20 ++++++----
.../apache/camel/component/feed/EntryFilter.java | 28 +++++++++++---
.../apache/camel/component/feed/FeedEndpoint.java | 35 ++---------------
.../component/feed/FeedEntryPollingConsumer.java | 28 ++++++++------
.../atom/AtomEntryPollingConsumerTest.java | 35 +++--------------
.../AtomEntryPollingConsumerWithBasicAuthTest.java | 4 --
...=> AtomEntryPollingConsumerWithResumeTest.java} | 45 +++++++++-------------
.../component/atom/UpdatedDateFilterTest.java | 2 +-
.../camel/component/rss/RssEndpointConfigurer.java | 9 -----
.../camel/component/rss/RssEndpointUriFactory.java | 4 +-
.../org/apache/camel/component/rss/rss.json | 2 -
.../apache/camel/component/rss/RssEndpoint.java | 6 +--
.../component/rss/RssEntryPollingConsumer.java | 14 ++-----
.../camel/component/rss/UpdatedDateFilter.java | 18 ++++++---
.../camel/component/rss/UpdatedDateFilterTest.java | 7 ++--
.../docs/modules/eips/pages/resume-strategies.adoc | 2 +
.../ROOT/pages/camel-3x-upgrade-guide-3_17.adoc | 12 ++++++
22 files changed, 118 insertions(+), 186 deletions(-)
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 ac9c85c7580..ad131f4cc5c 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
@@ -36,12 +36,9 @@ public class AtomEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
case "feedheader":
case "feedHeader": target.setFeedHeader(property(camelContext, boolean.class, value)); return true;
- case "filter": target.setFilter(property(camelContext, boolean.class, value)); return true;
case "greedy": target.setGreedy(property(camelContext, boolean.class, value)); return true;
case "initialdelay":
case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true;
- case "lastupdate":
- case "lastUpdate": target.setLastUpdate(property(camelContext, java.util.Date.class, value)); return true;
case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
case "pollstrategy":
case "pollStrategy": target.setPollStrategy(property(camelContext, org.apache.camel.spi.PollingConsumerPollStrategy.class, value)); return true;
@@ -91,12 +88,9 @@ public class AtomEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": return org.apache.camel.ExchangePattern.class;
case "feedheader":
case "feedHeader": return boolean.class;
- case "filter": return boolean.class;
case "greedy": return boolean.class;
case "initialdelay":
case "initialDelay": return long.class;
- case "lastupdate":
- case "lastUpdate": return java.util.Date.class;
case "password": return java.lang.String.class;
case "pollstrategy":
case "pollStrategy": return org.apache.camel.spi.PollingConsumerPollStrategy.class;
@@ -147,12 +141,9 @@ public class AtomEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": return target.getExchangePattern();
case "feedheader":
case "feedHeader": return target.isFeedHeader();
- case "filter": return target.isFilter();
case "greedy": return target.isGreedy();
case "initialdelay":
case "initialDelay": return target.getInitialDelay();
- case "lastupdate":
- case "lastUpdate": return target.getLastUpdate();
case "password": return target.getPassword();
case "pollstrategy":
case "pollStrategy": return target.getPollStrategy();
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 8e236eaab1c..ee3c0587665 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
@@ -21,7 +21,7 @@ public class AtomEndpointUriFactory extends org.apache.camel.support.component.E
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(28);
+ Set<String> props = new HashSet<>(26);
props.add("backoffErrorThreshold");
props.add("backoffIdleThreshold");
props.add("backoffMultiplier");
@@ -31,10 +31,8 @@ public class AtomEndpointUriFactory extends org.apache.camel.support.component.E
props.add("exchangePattern");
props.add("feedHeader");
props.add("feedUri");
- props.add("filter");
props.add("greedy");
props.add("initialDelay");
- props.add("lastUpdate");
props.add("password");
props.add("pollStrategy");
props.add("repeatCount");
diff --git a/components/camel-atom/src/generated/resources/org/apache/camel/component/atom/atom.json b/components/camel-atom/src/generated/resources/org/apache/camel/component/atom/atom.json
index c2bbee67d88..f12f6940fbb 100644
--- a/components/camel-atom/src/generated/resources/org/apache/camel/component/atom/atom.json
+++ b/components/camel-atom/src/generated/resources/org/apache/camel/component/atom/atom.json
@@ -32,8 +32,6 @@
"feedUri": { "kind": "path", "displayName": "Feed Uri", "group": "consumer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The URI to the feed to poll." },
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a m [...]
"feedHeader": { "kind": "parameter", "displayName": "Feed Header", "group": "consumer", "label": "", "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." },
- "filter": { "kind": "parameter", "displayName": "Filter", "group": "consumer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Sets whether to use filtering or not of the entries." },
- "lastUpdate": { "kind": "parameter", "displayName": "Last Update", "group": "consumer", "label": "", "required": false, "type": "string", "javaType": "java.util.Date", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the timestamp to be used for filtering entries from the atom feeds. This options is only in conjunction with the splitEntries." },
"password": { "kind": "parameter", "displayName": "Password", "group": "consumer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the password to be used for basic authentication when polling from a HTTP feed." },
"sendEmptyMessageWhenIdle": { "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
"sortEntries": { "kind": "parameter", "displayName": "Sort Entries", "group": "consumer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether to sort entries by published date. Only works when splitEntries = true." },
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 ec7f71b8e4a..c69b02365aa 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
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.atom;
-import java.util.Date;
-
import org.apache.abdera.model.Feed;
import org.apache.camel.Category;
import org.apache.camel.Exchange;
@@ -57,9 +55,9 @@ public class AtomEndpoint extends FeedEndpoint {
@Override
protected FeedPollingConsumer createEntryPollingConsumer(
- FeedEndpoint feedEndpoint, Processor processor, boolean filter, Date lastUpdate, boolean throttleEntries)
+ FeedEndpoint feedEndpoint, Processor processor, boolean throttleEntries)
throws Exception {
- AtomEntryPollingConsumer answer = new AtomEntryPollingConsumer(this, processor, filter, lastUpdate, throttleEntries);
+ AtomEntryPollingConsumer answer = new AtomEntryPollingConsumer(this, processor, throttleEntries);
configureConsumer(answer);
return answer;
}
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 aed7007a5f2..0d5b02ed291 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
@@ -17,13 +17,11 @@
package org.apache.camel.component.atom;
import java.io.IOException;
-import java.util.Date;
import org.apache.abdera.model.Document;
import org.apache.abdera.model.Feed;
import org.apache.abdera.parser.ParseException;
import org.apache.camel.Processor;
-import org.apache.camel.component.feed.EntryFilter;
import org.apache.camel.component.feed.FeedEntryPollingConsumer;
import org.apache.camel.util.ObjectHelper;
@@ -33,9 +31,8 @@ import org.apache.camel.util.ObjectHelper;
public class AtomEntryPollingConsumer extends FeedEntryPollingConsumer {
private Document<Feed> document;
- public AtomEntryPollingConsumer(AtomEndpoint endpoint, Processor processor, boolean filter, Date lastUpdate,
- boolean throttleEntries) {
- super(endpoint, processor, filter, lastUpdate, throttleEntries);
+ public AtomEntryPollingConsumer(AtomEndpoint endpoint, Processor processor, boolean throttleEntries) {
+ super(endpoint, processor, throttleEntries);
}
private Document<Feed> getDocument() throws IOException, ParseException {
@@ -73,9 +70,4 @@ public class AtomEntryPollingConsumer extends FeedEntryPollingConsumer {
protected void resetList() {
document = null;
}
-
- @Override
- protected EntryFilter createEntryFilter(Date lastUpdate) {
- return new UpdatedDateFilter(lastUpdate);
- }
}
diff --git a/components/camel-atom/src/main/java/org/apache/camel/component/atom/UpdatedDateFilter.java b/components/camel-atom/src/main/java/org/apache/camel/component/atom/UpdatedDateFilter.java
index 9dfa81f6386..1e4e69f23e3 100644
--- a/components/camel-atom/src/main/java/org/apache/camel/component/atom/UpdatedDateFilter.java
+++ b/components/camel-atom/src/main/java/org/apache/camel/component/atom/UpdatedDateFilter.java
@@ -20,7 +20,6 @@ import java.util.Date;
import org.apache.abdera.model.Entry;
import org.apache.camel.component.feed.EntryFilter;
-import org.apache.camel.component.feed.FeedEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,21 +27,29 @@ import org.slf4j.LoggerFactory;
* Filters out all entries which occur before the last time of the entry we saw (assuming entries arrive sorted in
* order).
*/
-public class UpdatedDateFilter implements EntryFilter {
+public class UpdatedDateFilter implements EntryFilter<Entry> {
private static final Logger LOG = LoggerFactory.getLogger(UpdatedDateFilter.class);
private Date lastUpdate;
+ public UpdatedDateFilter() {
+
+ }
+
public UpdatedDateFilter(Date lastUpdate) {
this.lastUpdate = lastUpdate;
}
+ private Date getLastUpdate() {
+ return this.lastUpdate;
+ }
+
@Override
- public boolean isValidEntry(FeedEndpoint endpoint, Object feed, Object entry) {
- Date updated = ((Entry) entry).getUpdated();
+ public boolean isValidEntry(Entry entry) {
+ Date updated = entry.getUpdated();
if (updated == null) {
// never been updated so get published date
- updated = ((Entry) entry).getPublished();
+ updated = entry.getPublished();
}
if (updated == null) {
LOG.debug("No updated time for entry so assuming its valid: entry=[{}]", entry);
@@ -51,12 +58,11 @@ public class UpdatedDateFilter implements EntryFilter {
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, entry);
+ LOG.debug("Entry is older than last update=[{}], no valid entry=[{}]", lastUpdate, entry);
return false;
}
}
lastUpdate = updated;
return true;
}
-
}
diff --git a/components/camel-atom/src/main/java/org/apache/camel/component/feed/EntryFilter.java b/components/camel-atom/src/main/java/org/apache/camel/component/feed/EntryFilter.java
index 436aefb3bf7..b477e1bb372 100644
--- a/components/camel-atom/src/main/java/org/apache/camel/component/feed/EntryFilter.java
+++ b/components/camel-atom/src/main/java/org/apache/camel/component/feed/EntryFilter.java
@@ -16,19 +16,35 @@
*/
package org.apache.camel.component.feed;
+import org.apache.camel.ResumeStrategy;
+
/**
* Filter used by the {@link org.apache.camel.component.feed.FeedEntryPollingConsumer} to filter entries from the feed.
+ *
+ * @param <E> entry type
*/
-public interface EntryFilter {
+public interface EntryFilter<E> extends ResumeStrategy {
/**
* Tests to be used as filtering the feed for only entries of interest, such as only new entries, etc.
*
- * @param endpoint the endpoint
- * @param feed the feed
- * @param entry the given entry to filter
- * @return <tt>true</tt> to include the entry, <ff>false</tt> to skip it
+ * @param entry the given entry to filter
+ * @return <tt>true</tt> to include the entry, <ff>false</tt> to skip it
*/
- boolean isValidEntry(FeedEndpoint endpoint, Object feed, Object entry);
+ boolean isValidEntry(E entry);
+
+ @Override
+ default void resume() {
+ // NO-OP by default. Implementations can implement more complex behaviors if needed
+ }
+
+ @Override
+ default void start() {
+ // NO-OP
+ }
+ @Override
+ default void stop() {
+ // NO-OP
+ }
}
diff --git a/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEndpoint.java b/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEndpoint.java
index fab4f49a7b5..69e41aabac0 100644
--- a/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEndpoint.java
+++ b/components/camel-atom/src/main/java/org/apache/camel/component/feed/FeedEndpoint.java
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.feed;
-import java.util.Date;
-
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -39,11 +37,7 @@ public abstract class FeedEndpoint extends DefaultPollingEndpoint {
@UriParam(defaultValue = "true", description = "Sets whether or not entries should be sent "
+ "individually or whether the entire feed should be sent as a single message")
protected boolean splitEntries = true;
- @UriParam(description = "Sets the timestamp to be used for filtering entries from the "
- + "atom feeds. This options is only in conjunction with the splitEntries.")
- protected Date lastUpdate;
- @UriParam(defaultValue = "true", description = "Sets whether to use filtering or not of the entries.")
- protected boolean filter = true;
+
@UriParam(defaultValue = "true", description = "Sets whether to add the feed object as a header.")
private boolean feedHeader = true;
@UriParam(description = "Sets whether to sort entries by published date. Only works when splitEntries = true.")
@@ -76,7 +70,7 @@ public abstract class FeedEndpoint extends DefaultPollingEndpoint {
FeedPollingConsumer answer;
if (isSplitEntries()) {
- answer = createEntryPollingConsumer(this, processor, filter, lastUpdate, throttleEntries);
+ answer = createEntryPollingConsumer(this, processor, throttleEntries);
} else {
answer = createPollingConsumer(this, processor);
}
@@ -92,7 +86,7 @@ public abstract class FeedEndpoint extends DefaultPollingEndpoint {
throws Exception;
protected abstract FeedPollingConsumer createEntryPollingConsumer(
- FeedEndpoint feedEndpoint, Processor processor, boolean filter, Date lastUpdate, boolean throttleEntries)
+ FeedEndpoint feedEndpoint, Processor processor, boolean throttleEntries)
throws Exception;
protected Exchange createExchangeWithFeedHeader(Object feed, String header) {
@@ -157,29 +151,6 @@ public abstract class FeedEndpoint extends DefaultPollingEndpoint {
this.splitEntries = splitEntries;
}
- public Date getLastUpdate() {
- return lastUpdate;
- }
-
- /**
- * Sets the timestamp to be used for filtering entries from the atom feeds. This options is only in conjunction with
- * the splitEntries.
- */
- public void setLastUpdate(Date lastUpdate) {
- this.lastUpdate = lastUpdate;
- }
-
- public boolean isFilter() {
- return filter;
- }
-
- /**
- * Sets whether to use filtering or not of the entries.
- */
- public void setFilter(boolean filter) {
- this.filter = filter;
- }
-
/**
* Sets whether to add the feed object as a header
*/
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 bd177f29044..d8bcfcf255e 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
@@ -16,29 +16,25 @@
*/
package org.apache.camel.component.feed;
-import java.util.Date;
import java.util.List;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.ResumeAware;
/**
* Consumer to poll feeds and return each entry from the feed step by step.
*/
-public abstract class FeedEntryPollingConsumer extends FeedPollingConsumer {
+public abstract class FeedEntryPollingConsumer<E> extends FeedPollingConsumer implements ResumeAware<EntryFilter<E>> {
protected int entryIndex;
- protected EntryFilter entryFilter;
+ protected EntryFilter<E> entryFilter;
@SuppressWarnings("rawtypes")
- protected List list;
+ protected List<E> list;
protected boolean throttleEntries;
protected Object feed;
- public FeedEntryPollingConsumer(FeedEndpoint endpoint, Processor processor, boolean filter, Date lastUpdate,
- boolean throttleEntries) {
+ public FeedEntryPollingConsumer(FeedEndpoint endpoint, Processor processor, boolean throttleEntries) {
super(endpoint, processor);
- if (filter) {
- entryFilter = createEntryFilter(lastUpdate);
- }
this.throttleEntries = throttleEntries;
}
@@ -52,12 +48,12 @@ public abstract class FeedEntryPollingConsumer extends FeedPollingConsumer {
int polledMessages = 0;
while (hasNextEntry()) {
- Object entry = list.get(entryIndex--);
+ E entry = list.get(entryIndex--);
polledMessages++;
boolean valid = true;
if (entryFilter != null) {
- valid = entryFilter.isValidEntry(endpoint, feed, entry);
+ valid = entryFilter.isValidEntry(entry);
}
if (valid) {
Exchange exchange = endpoint.createExchange(feed, entry);
@@ -76,7 +72,15 @@ public abstract class FeedEntryPollingConsumer extends FeedPollingConsumer {
return polledMessages;
}
- protected abstract EntryFilter createEntryFilter(Date lastUpdate);
+ @Override
+ public void setResumeStrategy(EntryFilter<E> resumeStrategy) {
+ this.entryFilter = resumeStrategy;
+ }
+
+ @Override
+ public EntryFilter<E> getResumeStrategy() {
+ return entryFilter;
+ }
protected abstract void resetList();
diff --git a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java
index c09517abb9e..25c24046be7 100644
--- a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java
+++ b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java
@@ -16,12 +16,10 @@
*/
package org.apache.camel.component.atom;
-import java.text.SimpleDateFormat;
-
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.spi.Registry;
import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
@@ -32,43 +30,20 @@ import org.junit.jupiter.api.condition.OS;
@DisabledOnOs(OS.AIX)
public class AtomEntryPollingConsumerTest extends CamelTestSupport {
+ @DisplayName("Tests whether can consume from the atom feed")
@Test
- void testResult() throws Exception {
+ void testCanConsume() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result1");
mock.expectedMessageCount(7);
mock.assertIsSatisfied();
}
- @Test
- void testResult2() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result2");
- mock.expectedMessageCount(7);
- mock.assertIsSatisfied();
- }
-
- @Test
- void testResult3() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result3");
- mock.expectedMessageCount(4);
- mock.assertIsSatisfied();
- }
-
- @Override
- protected void bindToRegistry(Registry registry) throws Exception {
- SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss Z");
- registry.bind("myDate", df.parse("2007-11-13 14:35:00 +0100"));
- }
-
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
- from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=500").to("mock:result1");
-
- from("atom:file:src/test/data/feed.atom?splitEntries=true&filter=false&delay=500").to("mock:result2");
-
- from("atom:file:src/test/data/feed.atom?splitEntries=true&filter=true&lastUpdate=#myDate&delay=500")
- .to("mock:result3");
+ from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=500")
+ .to("mock:result1");
}
};
}
diff --git a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithBasicAuthTest.java b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithBasicAuthTest.java
index cd4fbc5d9bc..a9f2778ae67 100644
--- a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithBasicAuthTest.java
+++ b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithBasicAuthTest.java
@@ -37,10 +37,6 @@ public class AtomEntryPollingConsumerWithBasicAuthTest extends AtomEntryPollingC
from("atom:http://localhost:" + JettyTestServer.getInstance().port
+ "/?splitEntries=true&filter=false&delay=500&username=camel&password=camelPass")
.to("mock:result2");
-
- from("atom:http://localhost:" + JettyTestServer.getInstance().port
- + "/?splitEntries=true&filter=true&lastUpdate=#myDate&delay=500&username=camel&password=camelPass")
- .to("mock:result3");
}
};
}
diff --git a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithResumeTest.java
similarity index 60%
copy from components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java
copy to components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithResumeTest.java
index c09517abb9e..b39e4d957c7 100644
--- a/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerTest.java
+++ b/components/camel-atom/src/test/java/org/apache/camel/component/atom/AtomEntryPollingConsumerWithResumeTest.java
@@ -16,12 +16,14 @@
*/
package org.apache.camel.component.atom;
+import java.text.ParseException;
import java.text.SimpleDateFormat;
+import java.util.Date;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.spi.Registry;
import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
@@ -30,45 +32,34 @@ import org.junit.jupiter.api.condition.OS;
* Unit test for AtomEntryPollingConsumer
*/
@DisabledOnOs(OS.AIX)
-public class AtomEntryPollingConsumerTest extends CamelTestSupport {
+public class AtomEntryPollingConsumerWithResumeTest extends CamelTestSupport {
+ @DisplayName("Tests whether the component can consume with resume API enabled")
@Test
- void testResult() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result1");
- mock.expectedMessageCount(7);
- mock.assertIsSatisfied();
- }
-
- @Test
- void testResult2() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result2");
- mock.expectedMessageCount(7);
- mock.assertIsSatisfied();
- }
-
- @Test
- void testResult3() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result3");
+ void testCanConsumeWithResume() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMessageCount(4);
mock.assertIsSatisfied();
}
- @Override
- protected void bindToRegistry(Registry registry) throws Exception {
+ private Date getTestFilterDate() {
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss Z");
- registry.bind("myDate", df.parse("2007-11-13 14:35:00 +0100"));
+
+ try {
+ return df.parse("2007-11-13 14:35:00 +0100");
+ } catch (ParseException e) {
+ throw new RuntimeException(e);
+ }
}
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
- from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=500").to("mock:result1");
-
- from("atom:file:src/test/data/feed.atom?splitEntries=true&filter=false&delay=500").to("mock:result2");
-
- from("atom:file:src/test/data/feed.atom?splitEntries=true&filter=true&lastUpdate=#myDate&delay=500")
- .to("mock:result3");
+ from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=500")
+ .resumable().resumeStrategy(new UpdatedDateFilter(getTestFilterDate()))
+ .routeId("WithResume")
+ .to("mock:result");
}
};
}
diff --git a/components/camel-atom/src/test/java/org/apache/camel/component/atom/UpdatedDateFilterTest.java b/components/camel-atom/src/test/java/org/apache/camel/component/atom/UpdatedDateFilterTest.java
index 7e6b2a64a14..d6147ab832c 100644
--- a/components/camel-atom/src/test/java/org/apache/camel/component/atom/UpdatedDateFilterTest.java
+++ b/components/camel-atom/src/test/java/org/apache/camel/component/atom/UpdatedDateFilterTest.java
@@ -53,7 +53,7 @@ public class UpdatedDateFilterTest {
// must reverse backwards
for (int i = entries.size() - 1; i > 0; i--) {
Entry entry = entries.get(i);
- boolean valid = filter.isValidEntry(null, doc, entry);
+ boolean valid = filter.isValidEntry(entry);
// only the 3 last should be true
if (i > 3) {
assertEquals(false, valid, "not valid");
diff --git a/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointConfigurer.java b/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointConfigurer.java
index f00c7c4480d..e874ec456a6 100644
--- a/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointConfigurer.java
+++ b/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointConfigurer.java
@@ -36,12 +36,9 @@ public class RssEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
case "feedheader":
case "feedHeader": target.setFeedHeader(property(camelContext, boolean.class, value)); return true;
- case "filter": target.setFilter(property(camelContext, boolean.class, value)); return true;
case "greedy": target.setGreedy(property(camelContext, boolean.class, value)); return true;
case "initialdelay":
case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true;
- case "lastupdate":
- case "lastUpdate": target.setLastUpdate(property(camelContext, java.util.Date.class, value)); return true;
case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
case "pollstrategy":
case "pollStrategy": target.setPollStrategy(property(camelContext, org.apache.camel.spi.PollingConsumerPollStrategy.class, value)); return true;
@@ -91,12 +88,9 @@ public class RssEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": return org.apache.camel.ExchangePattern.class;
case "feedheader":
case "feedHeader": return boolean.class;
- case "filter": return boolean.class;
case "greedy": return boolean.class;
case "initialdelay":
case "initialDelay": return long.class;
- case "lastupdate":
- case "lastUpdate": return java.util.Date.class;
case "password": return java.lang.String.class;
case "pollstrategy":
case "pollStrategy": return org.apache.camel.spi.PollingConsumerPollStrategy.class;
@@ -147,12 +141,9 @@ public class RssEndpointConfigurer extends PropertyConfigurerSupport implements
case "exchangePattern": return target.getExchangePattern();
case "feedheader":
case "feedHeader": return target.isFeedHeader();
- case "filter": return target.isFilter();
case "greedy": return target.isGreedy();
case "initialdelay":
case "initialDelay": return target.getInitialDelay();
- case "lastupdate":
- case "lastUpdate": return target.getLastUpdate();
case "password": return target.getPassword();
case "pollstrategy":
case "pollStrategy": return target.getPollStrategy();
diff --git a/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointUriFactory.java b/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointUriFactory.java
index ef51db65811..d539f17b1f6 100644
--- a/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointUriFactory.java
+++ b/components/camel-rss/src/generated/java/org/apache/camel/component/rss/RssEndpointUriFactory.java
@@ -21,7 +21,7 @@ public class RssEndpointUriFactory extends org.apache.camel.support.component.En
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(28);
+ Set<String> props = new HashSet<>(26);
props.add("backoffErrorThreshold");
props.add("backoffIdleThreshold");
props.add("backoffMultiplier");
@@ -31,10 +31,8 @@ public class RssEndpointUriFactory extends org.apache.camel.support.component.En
props.add("exchangePattern");
props.add("feedHeader");
props.add("feedUri");
- props.add("filter");
props.add("greedy");
props.add("initialDelay");
- props.add("lastUpdate");
props.add("password");
props.add("pollStrategy");
props.add("repeatCount");
diff --git a/components/camel-rss/src/generated/resources/org/apache/camel/component/rss/rss.json b/components/camel-rss/src/generated/resources/org/apache/camel/component/rss/rss.json
index 53abc7e1f60..e4ab08bf7d0 100644
--- a/components/camel-rss/src/generated/resources/org/apache/camel/component/rss/rss.json
+++ b/components/camel-rss/src/generated/resources/org/apache/camel/component/rss/rss.json
@@ -32,8 +32,6 @@
"feedUri": { "kind": "path", "displayName": "Feed Uri", "group": "consumer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The URI to the feed to poll." },
"bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a m [...]
"feedHeader": { "kind": "parameter", "displayName": "Feed Header", "group": "consumer", "label": "", "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." },
- "filter": { "kind": "parameter", "displayName": "Filter", "group": "consumer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Sets whether to use filtering or not of the entries." },
- "lastUpdate": { "kind": "parameter", "displayName": "Last Update", "group": "consumer", "label": "", "required": false, "type": "string", "javaType": "java.util.Date", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the timestamp to be used for filtering entries from the atom feeds. This options is only in conjunction with the splitEntries." },
"password": { "kind": "parameter", "displayName": "Password", "group": "consumer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the password to be used for basic authentication when polling from a HTTP feed." },
"sendEmptyMessageWhenIdle": { "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
"sortEntries": { "kind": "parameter", "displayName": "Sort Entries", "group": "consumer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether to sort entries by published date. Only works when splitEntries = true." },
diff --git a/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEndpoint.java b/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEndpoint.java
index c70f395e64f..93e5857da33 100644
--- a/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEndpoint.java
+++ b/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEndpoint.java
@@ -17,7 +17,6 @@
package org.apache.camel.component.rss;
import java.util.Arrays;
-import java.util.Date;
import com.rometools.rome.feed.synd.SyndEntry;
import com.rometools.rome.feed.synd.SyndFeed;
@@ -77,10 +76,9 @@ public class RssEndpoint extends FeedEndpoint {
@Override
protected FeedPollingConsumer createEntryPollingConsumer(
- FeedEndpoint feedEndpoint, Processor processor,
- boolean filter, Date lastUpdate, boolean throttleEntries)
+ FeedEndpoint feedEndpoint, Processor processor, boolean throttleEntries)
throws Exception {
- RssEntryPollingConsumer answer = new RssEntryPollingConsumer(this, processor, filter, lastUpdate, throttleEntries);
+ RssEntryPollingConsumer answer = new RssEntryPollingConsumer(this, processor, throttleEntries);
configureConsumer(answer);
return answer;
}
diff --git a/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEntryPollingConsumer.java b/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEntryPollingConsumer.java
index cff7b885c24..3041a25c651 100644
--- a/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEntryPollingConsumer.java
+++ b/components/camel-rss/src/main/java/org/apache/camel/component/rss/RssEntryPollingConsumer.java
@@ -17,11 +17,9 @@
package org.apache.camel.component.rss;
import java.util.Collections;
-import java.util.Date;
import com.rometools.rome.feed.synd.SyndFeed;
import org.apache.camel.Processor;
-import org.apache.camel.component.feed.EntryFilter;
import org.apache.camel.component.feed.FeedEntryPollingConsumer;
import org.apache.camel.util.ObjectHelper;
@@ -30,13 +28,12 @@ import org.apache.camel.util.ObjectHelper;
*/
public class RssEntryPollingConsumer extends FeedEntryPollingConsumer {
- public RssEntryPollingConsumer(RssEndpoint endpoint, Processor processor, boolean filter, Date lastUpdate,
- boolean throttleEntries) {
- super(endpoint, processor, filter, lastUpdate, throttleEntries);
+ public RssEntryPollingConsumer(RssEndpoint endpoint, Processor processor, boolean throttleEntries) {
+ super(endpoint, processor, throttleEntries);
}
@Override
- protected void populateList(Object feed) throws Exception {
+ protected void populateList(Object feed) {
if (list == null) {
list = ((SyndFeed) feed).getEntries();
if (endpoint.isSortEntries()) {
@@ -65,9 +62,4 @@ public class RssEntryPollingConsumer extends FeedEntryPollingConsumer {
protected void resetList() {
list = null;
}
-
- @Override
- protected EntryFilter createEntryFilter(Date lastUpdate) {
- return new UpdatedDateFilter(lastUpdate);
- }
}
diff --git a/components/camel-rss/src/main/java/org/apache/camel/component/rss/UpdatedDateFilter.java b/components/camel-rss/src/main/java/org/apache/camel/component/rss/UpdatedDateFilter.java
index 1fc3a48ed0f..22dde13dc98 100644
--- a/components/camel-rss/src/main/java/org/apache/camel/component/rss/UpdatedDateFilter.java
+++ b/components/camel-rss/src/main/java/org/apache/camel/component/rss/UpdatedDateFilter.java
@@ -21,7 +21,6 @@ import java.util.Map;
import com.rometools.rome.feed.synd.SyndEntry;
import org.apache.camel.component.feed.EntryFilter;
-import org.apache.camel.component.feed.FeedEndpoint;
import org.apache.camel.support.LRUCacheFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,7 +29,7 @@ import org.slf4j.LoggerFactory;
* Filters out all entries which occur before the last time of the entry we saw (assuming entries arrive sorted in
* order).
*/
-public class UpdatedDateFilter implements EntryFilter {
+public class UpdatedDateFilter implements EntryFilter<SyndEntry> {
private static final Logger LOG = LoggerFactory.getLogger(UpdatedDateFilter.class);
private Date lastUpdate;
@@ -38,16 +37,23 @@ public class UpdatedDateFilter implements EntryFilter {
@SuppressWarnings("unchecked")
private Map<Integer, Integer> entriesForLastUpdate = LRUCacheFactory.newLRUCache(1000);
+ public UpdatedDateFilter() {
+ }
+
public UpdatedDateFilter(Date lastUpdate) {
this.lastUpdate = lastUpdate;
}
+ public Date getLastUpdate() {
+ return lastUpdate;
+ }
+
@Override
- public boolean isValidEntry(FeedEndpoint endpoint, Object feed, Object entry) {
- Date updated = ((SyndEntry) entry).getUpdatedDate();
+ public boolean isValidEntry(SyndEntry entry) {
+ Date updated = entry.getUpdatedDate();
if (updated == null) {
// never been updated so get published date
- updated = ((SyndEntry) entry).getPublishedDate();
+ updated = entry.getPublishedDate();
}
if (updated == null) {
LOG.debug("No updated time for entry so assuming its valid: entry=[{}]", entry);
@@ -55,7 +61,7 @@ public class UpdatedDateFilter implements EntryFilter {
}
if (lastUpdate != null) {
if (lastUpdate.after(updated)) {
- LOG.debug("Entry is older than lastupdate=[{}], no valid entry=[{}]", lastUpdate, entry);
+ LOG.debug("Entry is older than last update=[{}], no valid entry=[{}]", lastUpdate, entry);
return false;
} else {
Integer hash = entry.hashCode();
diff --git a/components/camel-rss/src/test/java/org/apache/camel/component/rss/UpdatedDateFilterTest.java b/components/camel-rss/src/test/java/org/apache/camel/component/rss/UpdatedDateFilterTest.java
index 08d1d5b7878..23f1123b662 100644
--- a/components/camel-rss/src/test/java/org/apache/camel/component/rss/UpdatedDateFilterTest.java
+++ b/components/camel-rss/src/test/java/org/apache/camel/component/rss/UpdatedDateFilterTest.java
@@ -20,7 +20,6 @@ import java.util.Date;
import com.rometools.rome.feed.synd.SyndEntry;
import com.rometools.rome.feed.synd.SyndEntryImpl;
-import com.rometools.rome.feed.synd.SyndFeedImpl;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -43,17 +42,17 @@ public class UpdatedDateFilterTest {
SyndEntry entry = new SyndEntryImpl();
entry.setPublishedDate(now);
entry.setAuthor("ANDY");
- assertTrue(fixture.isValidEntry(new RssEndpoint(), new SyndFeedImpl(), entry));
+ assertTrue(fixture.isValidEntry(entry));
entry = new SyndEntryImpl();
entry.setPublishedDate(now);
entry.setAuthor("ANDY");
- assertFalse(fixture.isValidEntry(new RssEndpoint(), new SyndFeedImpl(), entry));
+ assertFalse(fixture.isValidEntry(entry));
entry = new SyndEntryImpl();
entry.setPublishedDate(now);
entry.setAuthor("FRED");
- assertTrue(fixture.isValidEntry(new RssEndpoint(), new SyndFeedImpl(), entry));
+ assertTrue(fixture.isValidEntry(entry));
}
}
diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc
index 606a1f813b5..11734e0e573 100644
--- a/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc
+++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/resume-strategies.adoc
@@ -11,10 +11,12 @@ The resume strategies can be used to allow quicker stop and resume operations wh
Support for resume varies according to the component. Initially, the support is available for the following components:
+* xref:components::atom-component.adoc[camel-atom]
* xref:components::aws2-kinesis-component.adoc[camel-aws2-kinesis]
* xref:components::couchdb-component.adoc[camel-couchdb]
* xref:components::file-component.adoc[camel-file]
* xref:components::kafka-component.adoc[camel-kafka]
+* xref:components::rss-component.adoc[camel-rss]
The resume strategies comes in 3 parts:
diff --git a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
index 0c98adc9d8d..61e118df0d2 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
@@ -101,6 +101,18 @@ Asynchronous, Synchronous or NO-OP commit policies from the former `autoCommitOn
The configuration for body handler file uploads has changed from `true` to `false`.
The configuration can be enabled via the `VertxPlatformHttpServerConfiguration` class.
+=== camel-atom
+
+This component was refactored to support the Resume API v2. As such, the options `filter` and `lastUpdate` where removed.
+
+```
+from("atom:file:src/test/data/feed.atom?splitEntries=true&delay=500")
+ .resumable().resumeStrategy(new UpdatedDateFilter(new Date()))
+ .to("mock:result");
+```
+
+More complex filters can be implemented by extending the `UpdatedDateFilter` or by implementing a new `EntryFilter` resume strategy.
+
=== camel-cdi
The support for the Camel XML configuration import, that had been marked as deprecated in previous releases, was removed.