Repository: camel
Updated Branches:
refs/heads/master 5067c5b89 -> a162ac8eb
CAMEL-10849: Salesforce: subscription channel ...
...created per component
Added option `replayId` to the Salesforce endpoint, this is taken in
preference of all other *replayId options. Now you can use `replayId`
parameter in the endpoint uri, e.g:
from("salesforce:AccountUpdates?replayId=-1")
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a162ac8e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a162ac8e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a162ac8e
Branch: refs/heads/master
Commit: a162ac8eb7003f52c9fcbd7eebd71c92676de6e2
Parents: 5067c5b
Author: Zoran Regvart <[email protected]>
Authored: Fri Feb 24 10:23:37 2017 +0100
Committer: Zoran Regvart <[email protected]>
Committed: Fri Feb 24 10:39:54 2017 +0100
----------------------------------------------------------------------
.../src/main/docs/salesforce-component.adoc | 3 ++-
.../camel/component/salesforce/SalesforceEndpoint.java | 11 +++++++++++
.../internal/streaming/SubscriptionHelper.java | 4 +++-
.../internal/streaming/SubscriptionHelperTest.java | 11 +++++++++++
4 files changed, 27 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a162ac8e/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
----------------------------------------------------------------------
diff --git
a/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
b/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
index 58279e6..c6c9aa2 100644
---
a/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
+++
b/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
@@ -468,7 +468,7 @@ The Salesforce component supports 23 options which are
listed below.
// endpoint options: START
-The Salesforce component supports 42 endpoint options which are listed below:
+The Salesforce component supports 43 endpoint options which are listed below:
{% raw %}
[width="100%",cols="2,1,1m,1m,5",options="header"]
@@ -513,6 +513,7 @@ The Salesforce component supports 42 endpoint options which
are listed below:
| sObjectSearch | common | | String | Salesforce SOSL search string
| updateTopic | common | false | boolean | Whether to update an existing Push
Topic when using the Streaming API defaults to false
| bridgeErrorHandler | consumer | false | boolean | 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 message and handled by the routing Error Handler. By default
the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with
exceptions that will be logged at WARN or ERROR level and ignored.
+| replayId | consumer | | Long | The replayId value to use when subscribing
| exceptionHandler | consumer (advanced) | | ExceptionHandler | To let the
consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler
is enabled then this options is not in use. By default the consumer will deal
with exceptions that will be logged at WARN or ERROR level and ignored.
| exchangePattern | consumer (advanced) | | ExchangePattern | Sets the
exchange pattern when the consumer creates an exchange.
| synchronous | advanced | false | boolean | Sets whether synchronous
processing should be strictly used or Camel is allowed to use asynchronous
processing (if supported).
http://git-wip-us.apache.org/repos/asf/camel/blob/a162ac8e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpoint.java
----------------------------------------------------------------------
diff --git
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpoint.java
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpoint.java
index 4b541c5..36ae4e3 100644
---
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpoint.java
+++
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpoint.java
@@ -45,6 +45,9 @@ public class SalesforceEndpoint extends DefaultEndpoint {
@UriParam
private final SalesforceEndpointConfig config;
+ @UriParam(label = "consumer", description = "The replayId value to use
when subscribing")
+ private Long replayId;
+
public SalesforceEndpoint(String uri, SalesforceComponent
salesforceComponent,
SalesforceEndpointConfig config, OperationName
operationName, String topicName) {
super(uri, salesforceComponent);
@@ -104,6 +107,14 @@ public class SalesforceEndpoint extends DefaultEndpoint {
return topicName;
}
+ public void setReplayId(final Long replayId) {
+ this.replayId = replayId;
+ }
+
+ public Long getReplayId() {
+ return replayId;
+ }
+
@Override
protected void doStart() throws Exception {
try {
http://git-wip-us.apache.org/repos/asf/camel/blob/a162ac8e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
----------------------------------------------------------------------
diff --git
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
index fb5d454..7212377 100644
---
a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
+++
b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
@@ -430,6 +430,8 @@ public class SubscriptionHelper extends ServiceSupport {
static Optional<Long> determineReplayIdFor(final SalesforceEndpoint
endpoint, final String topicName) {
final String channelName = getChannelName(topicName);
+ final Long replayId = endpoint.getReplayId();
+
final SalesforceComponent component = endpoint.getComponent();
final SalesforceEndpointConfig endpointConfiguration =
endpoint.getConfiguration();
@@ -444,7 +446,7 @@ public class SubscriptionHelper extends ServiceSupport {
// the endpoint values have priority over component values, and the
default values posteriority
// over give topic values
- return Stream.of(endpointReplayId, componentReplayId,
endpointDefaultReplayId, componentDefaultReplayId)
+ return Stream.of(replayId, endpointReplayId, componentReplayId,
endpointDefaultReplayId, componentDefaultReplayId)
.filter(Objects::nonNull).findFirst();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a162ac8e/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperTest.java
----------------------------------------------------------------------
diff --git
a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperTest.java
b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperTest.java
index 6e2d7c6..dc5e454 100644
---
a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperTest.java
+++
b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperTest.java
@@ -47,6 +47,7 @@ public class SubscriptionHelperTest {
final SalesforceComponent component = mock(SalesforceComponent.class);
final SalesforceEndpoint endpoint = mock(SalesforceEndpoint.class);
+ when(endpoint.getReplayId()).thenReturn(null);
when(endpoint.getComponent()).thenReturn(component);
when(endpoint.getConfiguration()).thenReturn(config);
when(component.getConfig()).thenReturn(new SalesforceEndpointConfig());
@@ -76,6 +77,7 @@ public class SubscriptionHelperTest {
when(component.getConfig()).thenReturn(componentConfig);
final SalesforceEndpoint endpoint = mock(SalesforceEndpoint.class);
+ when(endpoint.getReplayId()).thenReturn(null);
when(endpoint.getComponent()).thenReturn(component);
when(endpoint.getConfiguration()).thenReturn(endpointConfig);
@@ -92,5 +94,14 @@ public class SubscriptionHelperTest {
assertEquals("Expecting replayId for `my-topic-3` to be 1, as it is
component's default when endpoint does not have a default",
Optional.of(1L), determineReplayIdFor(endpoint,
"my-topic-3"));
+
+ when(endpoint.getReplayId()).thenReturn(6L);
+
+ assertEquals("Expecting replayId for `my-topic-1` to be 6, as it is
endpoint configured explicitly on the endpoint",
+ Optional.of(6L), determineReplayIdFor(endpoint,
"my-topic-1"));
+ assertEquals("Expecting replayId for `my-topic-2` to be 6, as it is
endpoint configured explicitly on the endpoint",
+ Optional.of(6L), determineReplayIdFor(endpoint,
"my-topic-2"));
+ assertEquals("Expecting replayId for `my-topic-3` to be 6, as it is
endpoint configured explicitly on the endpoint",
+ Optional.of(6L), determineReplayIdFor(endpoint,
"my-topic-3"));
}
}
