Repository: camel Updated Branches: refs/heads/master 9a5357e66 -> cdf97cb52
CAMEL-10849: Salesforce: subscription channel ... ...created per component Changes the `replayId` value from `Integer` to `Long`. The range of the `replayId` is explicitly specified in the Salesforce documentation it's safer to make it as big as can be. Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/cdf97cb5 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/cdf97cb5 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/cdf97cb5 Branch: refs/heads/master Commit: cdf97cb5223f555ea80fab5090afa1c167703e0a Parents: 9a5357e Author: Zoran Regvart <[email protected]> Authored: Wed Feb 22 17:49:03 2017 +0100 Committer: Zoran Regvart <[email protected]> Committed: Wed Feb 22 17:49:03 2017 +0100 ---------------------------------------------------------------------- .../src/main/docs/salesforce-component.adoc | 2 +- .../salesforce/SalesforceEndpointConfig.java | 12 +++---- .../streaming/CometDReplayExtension.java | 16 +++++++-- .../internal/streaming/SubscriptionHelper.java | 16 ++++----- .../streaming/SubscriptionHelperTest.java | 34 ++++++++++---------- .../SalesforceComponentConfiguration.java | 6 ++-- 6 files changed, 48 insertions(+), 38 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/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 fc102b9..58279e6 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 @@ -483,7 +483,7 @@ The Salesforce component supports 42 endpoint options which are listed below: | backoffIncrement | common | | long | Backoff interval increment for Streaming connection restart attempts for failures beyond CometD auto-reconnect. | batchId | common | | String | Bulk API Batch ID | contentType | common | | ContentType | Bulk API content type one of XML CSV ZIP_XML ZIP_CSV -| defaultReplayId | common | | Integer | Default replayId setting if no value is found in link initialReplayIdMap +| defaultReplayId | common | | Long | Default replayId setting if no value is found in link initialReplayIdMap | format | common | | PayloadFormat | Payload format to use for Salesforce API calls either JSON or XML defaults to JSON | httpClient | common | | SalesforceHttpClient | Custom Jetty Http Client to use to connect to Salesforce. | includeDetails | common | | Boolean | Include details in Salesforce1 Analytics report defaults to false. http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java ---------------------------------------------------------------------- diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java index d8b37a2..1aad7da 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java @@ -161,9 +161,9 @@ public class SalesforceEndpointConfig implements Cloneable { // Streaming API properties @UriParam - private Integer defaultReplayId; + private Long defaultReplayId; @UriParam - private Map<String, Integer> initialReplayIdMap; + private Map<String, Long> initialReplayIdMap; // Approval API properties private ApprovalRequest approval; @@ -615,7 +615,7 @@ public class SalesforceEndpointConfig implements Cloneable { return Collections.unmodifiableMap(valueMap); } - public Integer getDefaultReplayId() { + public Long getDefaultReplayId() { return defaultReplayId; } @@ -624,18 +624,18 @@ public class SalesforceEndpointConfig implements Cloneable { * * @param defaultReplayId */ - public void setDefaultReplayId(Integer defaultReplayId) { + public void setDefaultReplayId(Long defaultReplayId) { this.defaultReplayId = defaultReplayId; } - public Map<String, Integer> getInitialReplayIdMap() { + public Map<String, Long> getInitialReplayIdMap() { return Optional.ofNullable(initialReplayIdMap).orElse(Collections.emptyMap()); } /** * Replay IDs to start from per channel name. */ - public void setInitialReplayIdMap(Map<String, Integer> initialReplayIdMap) { + public void setInitialReplayIdMap(Map<String, Long> initialReplayIdMap) { this.initialReplayIdMap = initialReplayIdMap; } http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/CometDReplayExtension.java ---------------------------------------------------------------------- diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/CometDReplayExtension.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/CometDReplayExtension.java index e061dc4..3081804 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/CometDReplayExtension.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/CometDReplayExtension.java @@ -63,16 +63,26 @@ import org.cometd.bayeux.client.ClientSession.Extension.Adapter; */ public class CometDReplayExtension extends Adapter { private static final String EXTENSION_NAME = "replay"; - private final ConcurrentMap<String, Integer> dataMap = new ConcurrentHashMap<>(); + private final ConcurrentMap<String, Long> dataMap = new ConcurrentHashMap<>(); private final AtomicBoolean supported = new AtomicBoolean(); - public void addTopicReplayId(final String topicName, final int replayId) { + public void addTopicReplayId(final String topicName, final long replayId) { dataMap.put(topicName, replayId); } @Override public boolean rcv(ClientSession session, Message.Mutable message) { - Integer replayId = (Integer)message.get(EXTENSION_NAME); + final Object value = message.get(EXTENSION_NAME); + + final Long replayId; + if (value instanceof Long) { + replayId = (Long)value; + } else if (value instanceof Number) { + replayId = ((Number)value).longValue(); + } else { + replayId = null; + } + if (this.supported.get() && replayId != null) { try { dataMap.put(message.getChannel(), replayId); http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/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 d27f440..1f593cf 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 @@ -415,7 +415,7 @@ public class SubscriptionHelper extends ServiceSupport { void setupReplay(final SalesforceEndpoint endpoint) { final String topicName = endpoint.getTopicName(); - final Optional<Integer> replayId = determineReplayIdFor(endpoint, topicName); + final Optional<Long> replayId = determineReplayIdFor(endpoint, topicName); if (replayId.isPresent()) { final String channelName = getChannelName(topicName); @@ -423,20 +423,20 @@ public class SubscriptionHelper extends ServiceSupport { } } - static Optional<Integer> determineReplayIdFor(final SalesforceEndpoint endpoint, final String topicName) { + static Optional<Long> determineReplayIdFor(final SalesforceEndpoint endpoint, final String topicName) { final String channelName = getChannelName(topicName); final SalesforceComponent component = endpoint.getComponent(); final SalesforceEndpointConfig endpointConfiguration = endpoint.getConfiguration(); - final Map<String, Integer> endpointInitialReplayIdMap = endpointConfiguration.getInitialReplayIdMap(); - final Integer endpointReplayId = endpointInitialReplayIdMap.getOrDefault(topicName, endpointInitialReplayIdMap.get(channelName)); - final Integer endpointDefaultReplayId = endpointConfiguration.getDefaultReplayId(); + final Map<String, Long> endpointInitialReplayIdMap = endpointConfiguration.getInitialReplayIdMap(); + final Long endpointReplayId = endpointInitialReplayIdMap.getOrDefault(topicName, endpointInitialReplayIdMap.get(channelName)); + final Long endpointDefaultReplayId = endpointConfiguration.getDefaultReplayId(); final SalesforceEndpointConfig componentConfiguration = component.getConfig(); - final Map<String, Integer> componentInitialReplayIdMap = componentConfiguration.getInitialReplayIdMap(); - final Integer componentReplayId = componentInitialReplayIdMap.getOrDefault(topicName, componentInitialReplayIdMap.get(channelName)); - final Integer componentDefaultReplayId = componentConfiguration.getDefaultReplayId(); + final Map<String, Long> componentInitialReplayIdMap = componentConfiguration.getInitialReplayIdMap(); + final Long componentReplayId = componentInitialReplayIdMap.getOrDefault(topicName, componentInitialReplayIdMap.get(channelName)); + final Long componentDefaultReplayId = componentConfiguration.getDefaultReplayId(); // the endpoint values have priority over component values, and the default values posteriority // over give topic values http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/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 860e255..6e2d7c6 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 @@ -35,13 +35,13 @@ public class SubscriptionHelperTest { @Test public void shouldSupportInitialConfigMapWithTwoKeySyntaxes() throws Exception { - final Map<String, Integer> initialReplayIdMap = new HashMap<>(); - initialReplayIdMap.put("my-topic-1", 10); - initialReplayIdMap.put("/topic/my-topic-1", 20); - initialReplayIdMap.put("/topic/my-topic-2", 30); + final Map<String, Long> initialReplayIdMap = new HashMap<>(); + initialReplayIdMap.put("my-topic-1", 10L); + initialReplayIdMap.put("/topic/my-topic-1", 20L); + initialReplayIdMap.put("/topic/my-topic-2", 30L); final SalesforceEndpointConfig config = new SalesforceEndpointConfig(); - config.setDefaultReplayId(14); + config.setDefaultReplayId(14L); config.setInitialReplayIdMap(initialReplayIdMap); final SalesforceComponent component = mock(SalesforceComponent.class); @@ -52,25 +52,25 @@ public class SubscriptionHelperTest { when(component.getConfig()).thenReturn(new SalesforceEndpointConfig()); assertEquals("Expecting replayId for `my-topic-1` to be 10, as short topic names have priority", - Optional.of(10), determineReplayIdFor(endpoint, "my-topic-1")); + Optional.of(10L), determineReplayIdFor(endpoint, "my-topic-1")); - assertEquals("Expecting replayId for `my-topic-2` to be 30, the only one given", Optional.of(30), + assertEquals("Expecting replayId for `my-topic-2` to be 30, the only one given", Optional.of(30L), determineReplayIdFor(endpoint, "my-topic-2")); - assertEquals("Expecting replayId for `my-topic-3` to be 14, the default", Optional.of(14), + assertEquals("Expecting replayId for `my-topic-3` to be 14, the default", Optional.of(14L), determineReplayIdFor(endpoint, "my-topic-3")); } @Test public void precedenceShouldBeFollowed() { final SalesforceEndpointConfig componentConfig = new SalesforceEndpointConfig(); - componentConfig.setDefaultReplayId(1); - componentConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-1", 2)); - componentConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-2", 3)); + componentConfig.setDefaultReplayId(1L); + componentConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-1", 2L)); + componentConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-2", 3L)); final SalesforceEndpointConfig endpointConfig = new SalesforceEndpointConfig(); - endpointConfig.setDefaultReplayId(4); - endpointConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-1", 5)); + endpointConfig.setDefaultReplayId(4L); + endpointConfig.setInitialReplayIdMap(Collections.singletonMap("my-topic-1", 5L)); final SalesforceComponent component = mock(SalesforceComponent.class); when(component.getConfig()).thenReturn(componentConfig); @@ -80,17 +80,17 @@ public class SubscriptionHelperTest { when(endpoint.getConfiguration()).thenReturn(endpointConfig); assertEquals("Expecting replayId for `my-topic-1` to be 5, as endpoint configuration has priority", - Optional.of(5), determineReplayIdFor(endpoint, "my-topic-1")); + Optional.of(5L), determineReplayIdFor(endpoint, "my-topic-1")); assertEquals("Expecting replayId for `my-topic-2` to be 3, as endpoint does not configure it", - Optional.of(3), determineReplayIdFor(endpoint, "my-topic-2")); + Optional.of(3L), determineReplayIdFor(endpoint, "my-topic-2")); assertEquals("Expecting replayId for `my-topic-3` to be 4, as it is endpoint's default", - Optional.of(4), determineReplayIdFor(endpoint, "my-topic-3")); + Optional.of(4L), determineReplayIdFor(endpoint, "my-topic-3")); endpointConfig.setDefaultReplayId(null); 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(1), determineReplayIdFor(endpoint, "my-topic-3")); + Optional.of(1L), determineReplayIdFor(endpoint, "my-topic-3")); } } http://git-wip-us.apache.org/repos/asf/camel/blob/cdf97cb5/platforms/spring-boot/components-starter/camel-salesforce-starter/src/main/java/org/apache/camel/component/salesforce/springboot/SalesforceComponentConfiguration.java ---------------------------------------------------------------------- diff --git a/platforms/spring-boot/components-starter/camel-salesforce-starter/src/main/java/org/apache/camel/component/salesforce/springboot/SalesforceComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-salesforce-starter/src/main/java/org/apache/camel/component/salesforce/springboot/SalesforceComponentConfiguration.java index 0713088..2946fb3 100644 --- a/platforms/spring-boot/components-starter/camel-salesforce-starter/src/main/java/org/apache/camel/component/salesforce/springboot/SalesforceComponentConfiguration.java +++ b/platforms/spring-boot/components-starter/camel-salesforce-starter/src/main/java/org/apache/camel/component/salesforce/springboot/SalesforceComponentConfiguration.java @@ -574,7 +574,7 @@ public class SalesforceComponentConfiguration { * * @param defaultReplayId */ - private Integer defaultReplayId; + private Long defaultReplayId; /** * Replay IDs to start from per channel name. */ @@ -909,11 +909,11 @@ public class SalesforceComponentConfiguration { this.maxBackoff = maxBackoff; } - public Integer getDefaultReplayId() { + public Long getDefaultReplayId() { return defaultReplayId; } - public void setDefaultReplayId(Integer defaultReplayId) { + public void setDefaultReplayId(Long defaultReplayId) { this.defaultReplayId = defaultReplayId; }
