You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/10/27 07:40:27 UTC
[camel] branch master updated (f11b8f3 -> 52f5e7f)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from f11b8f3 Fixed CS.
new b139e82 CAMEL-11224 aws-sqs producer does not support new FIFO queues
new d2f53a6 addressed review comments
new 52f5e7f CAMEL-11224: Polished. This closes #2018
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../camel-aws/src/main/docs/aws-sqs-component.adoc | 4 +-
...ts.java => ConstantMessageGroupIdStrategy.java} | 21 +++---
... ExchangeIdMessageDeduplicationIdStrategy.java} | 21 +++---
....java => ExchangeIdMessageGroupIdStrategy.java} | 21 +++---
...ts.java => MessageDeduplicationIdStrategy.java} | 18 ++---
...sConstants.java => MessageGroupIdStrategy.java} | 18 ++---
...ava => NullMessageDeduplicationIdStrategy.java} | 21 +++---
...va => PropertyValueMessageGroupIdStrategy.java} | 21 +++---
.../camel/component/aws/sqs/SqsComponent.java | 5 ++
.../camel/component/aws/sqs/SqsConfiguration.java | 56 ++++++++++++++
.../camel/component/aws/sqs/SqsConstants.java | 1 +
.../camel/component/aws/sqs/SqsProducer.java | 35 ++++++---
.../component/aws/sqs/SqsConfigurationTest.java | 49 ++++++++++++
.../camel/component/aws/sqs/SqsProducerTest.java | 87 ++++++++++++++++++++++
14 files changed, 283 insertions(+), 95 deletions(-)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => ConstantMessageGroupIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => ExchangeIdMessageDeduplicationIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => ExchangeIdMessageGroupIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => MessageDeduplicationIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => MessageGroupIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => NullMessageDeduplicationIdStrategy.java} (67%)
copy components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/{SqsConstants.java => PropertyValueMessageGroupIdStrategy.java} (67%)
create mode 100644 components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
--
To stop receiving notification emails like this one, please contact
['"commits@camel.apache.org" <co...@camel.apache.org>'].
[camel] 01/03: CAMEL-11224 aws-sqs producer does not support new
FIFO queues
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit b139e822bcb79cf51ca5a3b9add5e6ef8e3a313d
Author: Jon McEwen <jo...@boohoo.com>
AuthorDate: Tue Oct 10 10:36:18 2017 +0100
CAMEL-11224 aws-sqs producer does not support new FIFO queues
---
...ts.java => ConstantMessageGroupIdStrategy.java} | 21 +++---
.../{SqsConstants.java => ExchangeIdStrategy.java} | 21 +++---
...{SqsConstants.java => HeaderValueStrategy.java} | 21 +++---
.../camel/component/aws/sqs/NullStrategy.java | 12 +++
.../camel/component/aws/sqs/SqsComponent.java | 4 +
.../camel/component/aws/sqs/SqsConfiguration.java | 56 ++++++++++++++
.../camel/component/aws/sqs/SqsConstants.java | 1 +
.../camel/component/aws/sqs/SqsProducer.java | 35 ++++++---
...s.java => StringValueFromExchangeStrategy.java} | 19 ++---
.../component/aws/sqs/SqsConfigurationTest.java | 32 ++++++++
.../camel/component/aws/sqs/SqsProducerTest.java | 87 ++++++++++++++++++++++
11 files changed, 251 insertions(+), 58 deletions(-)
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
similarity index 67%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
index 61891c9..44caf36 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
@@ -16,16 +16,13 @@
*/
package org.apache.camel.component.aws.sqs;
-/**
- * Constants used in Camel AWS SQS module
- *
- */
-public interface SqsConstants {
+import org.apache.camel.Exchange;
+
+public class ConstantMessageGroupIdStrategy implements StringValueFromExchangeStrategy {
+
+ @Override
+ public String value(Exchange exchange) {
+ return "CamelSingleMessageGroup";
+ }
- String ATTRIBUTES = "CamelAwsSqsAttributes";
- String MESSAGE_ATTRIBUTES = "CamelAwsSqsMessageAttributes";
- String MD5_OF_BODY = "CamelAwsSqsMD5OfBody";
- String MESSAGE_ID = "CamelAwsSqsMessageId";
- String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
- String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
-}
\ No newline at end of file
+}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java
similarity index 67%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java
index 61891c9..6b0dac1 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java
@@ -16,16 +16,13 @@
*/
package org.apache.camel.component.aws.sqs;
-/**
- * Constants used in Camel AWS SQS module
- *
- */
-public interface SqsConstants {
+import org.apache.camel.Exchange;
+
+public class ExchangeIdStrategy implements StringValueFromExchangeStrategy {
+
+ @Override
+ public String value(Exchange exchange) {
+ return exchange.getExchangeId();
+ }
- String ATTRIBUTES = "CamelAwsSqsAttributes";
- String MESSAGE_ATTRIBUTES = "CamelAwsSqsMessageAttributes";
- String MD5_OF_BODY = "CamelAwsSqsMD5OfBody";
- String MESSAGE_ID = "CamelAwsSqsMessageId";
- String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
- String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
-}
\ No newline at end of file
+}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java
similarity index 67%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java
index 61891c9..0bef8c2 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java
@@ -16,16 +16,13 @@
*/
package org.apache.camel.component.aws.sqs;
-/**
- * Constants used in Camel AWS SQS module
- *
- */
-public interface SqsConstants {
+import org.apache.camel.Exchange;
+
+public class HeaderValueStrategy implements StringValueFromExchangeStrategy {
+
+ @Override
+ public String value(Exchange exchange) {
+ return exchange.getIn().getHeader(SqsConstants.MESSAGE_GROUP_ID_HEADER, String.class);
+ }
- String ATTRIBUTES = "CamelAwsSqsAttributes";
- String MESSAGE_ATTRIBUTES = "CamelAwsSqsMessageAttributes";
- String MD5_OF_BODY = "CamelAwsSqsMD5OfBody";
- String MESSAGE_ID = "CamelAwsSqsMessageId";
- String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
- String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
-}
\ No newline at end of file
+}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java
new file mode 100644
index 0000000..bbb5ab3
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java
@@ -0,0 +1,12 @@
+package org.apache.camel.component.aws.sqs;
+
+import org.apache.camel.Exchange;
+
+public class NullStrategy implements StringValueFromExchangeStrategy {
+
+ @Override
+ public String value(Exchange exchange) {
+ return null;
+ }
+
+}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
index f865670..653b956 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
@@ -62,6 +62,10 @@ public class SqsComponent extends UriEndpointComponent {
throw new IllegalArgumentException("Extending message visibility (extendMessageVisibility) requires visibilityTimeout to be set on the Endpoint.");
}
+ if (configuration.isFifoQueue() && configuration.getMessageGroupIdStrategy() == null) {
+ throw new IllegalArgumentException("messageGroupIdStrategy must be set for FIFO queues.");
+ }
+
SqsEndpoint sqsEndpoint = new SqsEndpoint(uri, this, configuration);
sqsEndpoint.setConsumerProperties(parameters);
return sqsEndpoint;
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
index 3683e44..dbab657f 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
@@ -67,6 +67,10 @@ public class SqsConfiguration {
// producer properties
@UriParam(label = "producer")
private Integer delaySeconds;
+ @UriParam(label = "producer")
+ private StringValueFromExchangeStrategy messageGroupIdStrategy;
+ @UriParam(label = "producer", defaultValue = "useExchangeId")
+ private StringValueFromExchangeStrategy messageDeduplicationIdStrategy = new ExchangeIdStrategy();
// queue properties
@UriParam(label = "queue")
@@ -83,6 +87,18 @@ public class SqsConfiguration {
private String redrivePolicy;
/**
+ * Whether or not the queue is a FIFO queue
+ */
+ public boolean isFifoQueue() {
+ // AWS docs suggest this is valid derivation.
+ // FIFO queue names must end with .fifo, and standard queues cannot
+ if (queueName.endsWith(".fifo")) {
+ return true;
+ }
+ return false;
+ }
+
+ /**
* The region with which the AWS-SQS client wants to work with.
* Only works if Camel creates the AWS-SQS client, i.e., if you explicitly set amazonSQSClient,
* then this setting will have no effect. You would have to set it on the client you create directly
@@ -364,4 +380,44 @@ public class SqsConfiguration {
public void setProxyPort(Integer proxyPort) {
this.proxyPort = proxyPort;
}
+
+ /**
+ * Since *Camel 2.20*. Only for FIFO queues. Strategy for setting the messageGroupId on the message.
+ * Can be one of the following options: *useConstant*, *useExchangeId*, *useHeaderValue*.
+ * For the *useHeaderValue* option, the value of header "CamelAwsMessageGroupId" will be used.
+ */
+ public void setMessageGroupIdStrategy(String strategy) {
+ if ("useConstant".equalsIgnoreCase(strategy)) {
+ messageGroupIdStrategy = new ConstantMessageGroupIdStrategy();
+ } else if ("useExchangeId".equalsIgnoreCase(strategy)) {
+ messageGroupIdStrategy = new ExchangeIdStrategy();
+ } else if ("useHeaderValue".equalsIgnoreCase(strategy)) {
+ messageGroupIdStrategy = new HeaderValueStrategy();
+ } else {
+ throw new IllegalArgumentException("Unrecognised MessageGroupIdStrategy: " + strategy);
+ }
+ }
+
+ public StringValueFromExchangeStrategy getMessageGroupIdStrategy() {
+ return messageGroupIdStrategy;
+ }
+
+ public StringValueFromExchangeStrategy getMessageDeduplicationIdStrategy() {
+ return messageDeduplicationIdStrategy;
+ }
+
+ /**
+ * Since *Camel 2.20*. Only for FIFO queues. Strategy for setting the messageDeduplicationId on the message.
+ * Can be one of the following options: *useExchangeId*, *useContentBasedDeduplication*.
+ * For the *useContentBasedDeduplication* option, no messageDeduplicationId will be set on the message.
+ */
+ public void setMessageDeduplicationIdStrategy(String strategy) {
+ if ("useExchangeId".equalsIgnoreCase(strategy)) {
+ messageDeduplicationIdStrategy = new ExchangeIdStrategy();
+ } else if ("useContentBasedDeduplication".equalsIgnoreCase(strategy)) {
+ messageDeduplicationIdStrategy = new NullStrategy();
+ } else {
+ throw new IllegalArgumentException("Unrecognised MessageDeduplicationIdStrategy: " + strategy);
+ }
+ }
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
index 61891c9..5fc53c6 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
@@ -28,4 +28,5 @@ public interface SqsConstants {
String MESSAGE_ID = "CamelAwsSqsMessageId";
String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
+ String MESSAGE_GROUP_ID_HEADER = "CamelAwsMessageGroupId";
}
\ No newline at end of file
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
index 682d75e..e3d2f00 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
@@ -42,9 +42,9 @@ import static org.apache.camel.component.aws.common.AwsExchangeUtil.getMessageFo
*
*/
public class SqsProducer extends DefaultProducer {
-
+
private static final Logger LOG = LoggerFactory.getLogger(SqsProducer.class);
-
+
private transient String sqsProducerToString;
public SqsProducer(SqsEndpoint endpoint) throws NoFactoryAvailableException {
@@ -56,18 +56,33 @@ public class SqsProducer extends DefaultProducer {
SendMessageRequest request = new SendMessageRequest(getQueueUrl(), body);
request.setMessageAttributes(translateAttributes(exchange.getIn().getHeaders(), exchange));
addDelay(request, exchange);
+ configureFifoAttributes(request, exchange);
LOG.trace("Sending request [{}] from exchange [{}]...", request, exchange);
-
+
SendMessageResult result = getClient().sendMessage(request);
-
+
LOG.trace("Received result [{}]", result);
-
+
Message message = getMessageForResponse(exchange);
message.setHeader(SqsConstants.MESSAGE_ID, result.getMessageId());
message.setHeader(SqsConstants.MD5_OF_BODY, result.getMD5OfMessageBody());
}
+ private void configureFifoAttributes(SendMessageRequest request, Exchange exchange) {
+ if (getEndpoint().getConfiguration().isFifoQueue()) {
+ // use strategies
+ StringValueFromExchangeStrategy messageGroupIdStrategy = getEndpoint().getConfiguration().getMessageGroupIdStrategy();
+ String messageGroupId = messageGroupIdStrategy.value(exchange);
+ request.setMessageGroupId(messageGroupId);
+
+ StringValueFromExchangeStrategy messageDeduplicationIdStrategy = getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy();
+ String messageDeduplicationId = messageDeduplicationIdStrategy.value(exchange);
+ request.setMessageDeduplicationId(messageDeduplicationId);
+
+ }
+ }
+
private void addDelay(SendMessageRequest request, Exchange exchange) {
Integer headerValue = exchange.getIn().getHeader(SqsConstants.DELAY_HEADER, Integer.class);
Integer delayValue;
@@ -81,20 +96,20 @@ public class SqsProducer extends DefaultProducer {
LOG.trace("found delay: " + delayValue);
request.setDelaySeconds(delayValue == null ? Integer.valueOf(0) : delayValue);
}
-
+
protected AmazonSQS getClient() {
return getEndpoint().getClient();
}
-
+
protected String getQueueUrl() {
return getEndpoint().getQueueUrl();
}
-
+
@Override
public SqsEndpoint getEndpoint() {
return (SqsEndpoint) super.getEndpoint();
}
-
+
@Override
public String toString() {
if (sqsProducerToString == null) {
@@ -102,7 +117,7 @@ public class SqsProducer extends DefaultProducer {
}
return sqsProducerToString;
}
-
+
private Map<String, MessageAttributeValue> translateAttributes(Map<String, Object> headers, Exchange exchange) {
Map<String, MessageAttributeValue> result = new HashMap<String, MessageAttributeValue>();
HeaderFilterStrategy headerFilterStrategy = getEndpoint().getHeaderFilterStrategy();
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
similarity index 67%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
index 61891c9..31eafb6 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
@@ -16,16 +16,11 @@
*/
package org.apache.camel.component.aws.sqs;
-/**
- * Constants used in Camel AWS SQS module
- *
- */
-public interface SqsConstants {
+import org.apache.camel.Exchange;
+
+@FunctionalInterface
+public interface StringValueFromExchangeStrategy {
+
+ String value(Exchange exchange);
- String ATTRIBUTES = "CamelAwsSqsAttributes";
- String MESSAGE_ATTRIBUTES = "CamelAwsSqsMessageAttributes";
- String MD5_OF_BODY = "CamelAwsSqsMD5OfBody";
- String MESSAGE_ID = "CamelAwsSqsMessageId";
- String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
- String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
-}
\ No newline at end of file
+}
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
new file mode 100644
index 0000000..ace5173
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
@@ -0,0 +1,32 @@
+package org.apache.camel.component.aws.sqs;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+public class SqsConfigurationTest {
+
+ @Test
+ public void itReturnsAnInformativeErrorForBadMessageGroupIdStrategy() throws Exception {
+ SqsConfiguration sqsConfiguration = new SqsConfiguration();
+ try {
+ sqsConfiguration.setMessageGroupIdStrategy("useUnknownStrategy");
+ fail("Should have thrown exception");
+ } catch (Exception e) {
+ assertTrue("Bad error message: " + e.getMessage(), e.getMessage().startsWith("Unrecognised MessageGroupIdStrategy"));
+ }
+ }
+
+
+ @Test
+ public void itReturnsAnInformativeErrorForBadMessageDeduplicationIdStrategy() throws Exception {
+ SqsConfiguration sqsConfiguration = new SqsConfiguration();
+ try {
+ sqsConfiguration.setMessageDeduplicationIdStrategy("useUnknownStrategy");
+ fail("Should have thrown exception");
+ } catch (Exception e) {
+ assertTrue("Bad error message: " + e.getMessage(), e.getMessage().startsWith("Unrecognised MessageDeduplicationIdStrategy"));
+ }
+ }
+
+}
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
index bed350c..c1d24f1 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
@@ -55,6 +55,7 @@ public class SqsProducerTest {
private static final String SAMPLE_MESSAGE_HEADER_VALUE_3 = "heder_value_3";
private static final String SAMPLE_MESSAGE_HEADER_NAME_4 = "CamelHeader_1";
private static final String SAMPLE_MESSAGE_HEADER_VALUE_4 = "testValue";
+ private static final String SAMPLE_EXCHANGE_ID = "ID:whatever-the-hostname-is-32818-1506943497897-1:1:8:1:75939";
Exchange exchange = mock(Exchange.class, RETURNS_DEEP_STUBS);
@@ -75,12 +76,14 @@ public class SqsProducerTest {
sqsConfiguration = new SqsConfiguration();
HeaderFilterStrategy headerFilterStrategy = new SqsHeaderFilterStrategy();
sqsConfiguration.setDelaySeconds(Integer.valueOf(0));
+ sqsConfiguration.setQueueName("queueName");
when(sqsEndpoint.getClient()).thenReturn(amazonSQSClient);
when(sqsEndpoint.getConfiguration()).thenReturn(sqsConfiguration);
when(amazonSQSClient.sendMessage(any(SendMessageRequest.class))).thenReturn(sendMessageResult);
when(exchange.getOut()).thenReturn(outMessage);
when(exchange.getIn()).thenReturn(inMessage);
when(exchange.getPattern()).thenReturn(ExchangePattern.InOnly);
+ when(exchange.getExchangeId()).thenReturn(SAMPLE_EXCHANGE_ID);
when(inMessage.getBody(String.class)).thenReturn(SAMPLE_MESSAGE_BODY);
when(sqsEndpoint.getQueueUrl()).thenReturn(QUEUE_URL);
when(sqsEndpoint.getHeaderFilterStrategy()).thenReturn(headerFilterStrategy);
@@ -194,8 +197,92 @@ public class SqsProducerTest {
.getStringValue());
assertEquals(3, capture.getValue().getMessageAttributes().size());
}
+
+ @Test
+ public void itSetsMessageGroupIdUsingConstantStrategy() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useConstant");
+
+ underTest.process(exchange);
+
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertEquals("CamelSingleMessageGroup", capture.getValue().getMessageGroupId());
+
+ }
+ @Test
+ public void itSetsMessageGroupIdUsingExchangeIdStrategy() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useExchangeId");
+
+ underTest.process(exchange);
+
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertEquals(SAMPLE_EXCHANGE_ID, capture.getValue().getMessageGroupId());
+
+ }
+ @Test
+ public void itSetsMessageGroupIdUsingHeaderValueStrategy() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useHeaderValue");
+ when(inMessage.getHeader(SqsConstants.MESSAGE_GROUP_ID_HEADER, String.class)).thenReturn("my-group-id");
+
+ underTest.process(exchange);
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertEquals("my-group-id", capture.getValue().getMessageGroupId());
+
+ }
+
+ @Test
+ public void itSetsMessageDedpulicationIdUsingExchangeIdStrategy() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useConstant");
+ sqsConfiguration.setMessageDeduplicationIdStrategy("useExchangeId");
+
+ underTest.process(exchange);
+
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertEquals(SAMPLE_EXCHANGE_ID, capture.getValue().getMessageDeduplicationId());
+
+ }
+
+ @Test
+ public void itSetsMessageDedpulicationIdUsingExchangeIdStrategyAsDefault() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useConstant");
+
+ underTest.process(exchange);
+
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertEquals(SAMPLE_EXCHANGE_ID, capture.getValue().getMessageDeduplicationId());
+
+ }
+
+ @Test
+ public void itDoesNotSetMessageDedpulicationIdUsingContentBasedDeduplicationStrategy() throws Exception {
+ sqsConfiguration.setQueueName("queueName.fifo");
+ sqsConfiguration.setMessageGroupIdStrategy("useConstant");
+ sqsConfiguration.setMessageDeduplicationIdStrategy("useContentBasedDeduplication");
+
+ underTest.process(exchange);
+
+ ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+ verify(amazonSQSClient).sendMessage(capture.capture());
+
+ assertNull(capture.getValue().getMessageDeduplicationId());
+
+ }
}
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 02/03: addressed review comments
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit d2f53a6c21b76ff54d0a6e5113c38fb63e8f8f1f
Author: Jon McEwen <jo...@boohoo.com>
AuthorDate: Tue Oct 24 16:34:39 2017 +0100
addressed review comments
---
.../aws/sqs/ConstantMessageGroupIdStrategy.java | 4 ++--
... ExchangeIdMessageDeduplicationIdStrategy.java} | 6 +++---
....java => ExchangeIdMessageGroupIdStrategy.java} | 4 ++--
...gy.java => MessageDeduplicationIdStrategy.java} | 5 ++---
...geStrategy.java => MessageGroupIdStrategy.java} | 7 +++----
...ava => NullMessageDeduplicationIdStrategy.java} | 6 +++---
.../camel/component/aws/sqs/NullStrategy.java | 12 -----------
...va => PropertyValueMessageGroupIdStrategy.java} | 6 +++---
.../camel/component/aws/sqs/SqsComponent.java | 3 ++-
.../camel/component/aws/sqs/SqsConfiguration.java | 24 +++++++++++-----------
.../camel/component/aws/sqs/SqsConstants.java | 2 +-
.../camel/component/aws/sqs/SqsProducer.java | 8 ++++----
.../camel/component/aws/sqs/SqsProducerTest.java | 4 ++--
13 files changed, 39 insertions(+), 52 deletions(-)
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
index 44caf36..be5be47 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
@@ -18,10 +18,10 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-public class ConstantMessageGroupIdStrategy implements StringValueFromExchangeStrategy {
+public class ConstantMessageGroupIdStrategy implements MessageGroupIdStrategy {
@Override
- public String value(Exchange exchange) {
+ public String getMessageGroupId(Exchange exchange) {
return "CamelSingleMessageGroup";
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageDeduplicationIdStrategy.java
similarity index 80%
rename from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java
rename to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageDeduplicationIdStrategy.java
index 0bef8c2..6a50ffc 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/HeaderValueStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageDeduplicationIdStrategy.java
@@ -18,11 +18,11 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-public class HeaderValueStrategy implements StringValueFromExchangeStrategy {
+public class ExchangeIdMessageDeduplicationIdStrategy implements MessageDeduplicationIdStrategy {
@Override
- public String value(Exchange exchange) {
- return exchange.getIn().getHeader(SqsConstants.MESSAGE_GROUP_ID_HEADER, String.class);
+ public String getMessageDeduplicationId(Exchange exchange) {
+ return exchange.getExchangeId();
}
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageGroupIdStrategy.java
similarity index 87%
rename from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java
rename to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageGroupIdStrategy.java
index 6b0dac1..af38e10 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ExchangeIdMessageGroupIdStrategy.java
@@ -18,10 +18,10 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-public class ExchangeIdStrategy implements StringValueFromExchangeStrategy {
+public class ExchangeIdMessageGroupIdStrategy implements MessageGroupIdStrategy {
@Override
- public String value(Exchange exchange) {
+ public String getMessageGroupId(Exchange exchange) {
return exchange.getExchangeId();
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageDeduplicationIdStrategy.java
similarity index 89%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageDeduplicationIdStrategy.java
index 31eafb6..7f79ddf 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageDeduplicationIdStrategy.java
@@ -18,9 +18,8 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-@FunctionalInterface
-public interface StringValueFromExchangeStrategy {
+public interface MessageDeduplicationIdStrategy {
- String value(Exchange exchange);
+ String getMessageDeduplicationId(Exchange exchange);
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageGroupIdStrategy.java
similarity index 88%
rename from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
rename to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageGroupIdStrategy.java
index 31eafb6..23cb325 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/StringValueFromExchangeStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/MessageGroupIdStrategy.java
@@ -18,9 +18,8 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-@FunctionalInterface
-public interface StringValueFromExchangeStrategy {
-
- String value(Exchange exchange);
+public interface MessageGroupIdStrategy {
+
+ String getMessageGroupId(Exchange exchange);
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullMessageDeduplicationIdStrategy.java
similarity index 83%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullMessageDeduplicationIdStrategy.java
index 44caf36..6ab4ad4 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullMessageDeduplicationIdStrategy.java
@@ -18,11 +18,11 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-public class ConstantMessageGroupIdStrategy implements StringValueFromExchangeStrategy {
+public class NullMessageDeduplicationIdStrategy implements MessageDeduplicationIdStrategy {
@Override
- public String value(Exchange exchange) {
- return "CamelSingleMessageGroup";
+ public String getMessageDeduplicationId(Exchange exchange) {
+ return null;
}
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java
deleted file mode 100644
index bbb5ab3..0000000
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/NullStrategy.java
+++ /dev/null
@@ -1,12 +0,0 @@
-package org.apache.camel.component.aws.sqs;
-
-import org.apache.camel.Exchange;
-
-public class NullStrategy implements StringValueFromExchangeStrategy {
-
- @Override
- public String value(Exchange exchange) {
- return null;
- }
-
-}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/PropertyValueMessageGroupIdStrategy.java
similarity index 79%
copy from components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
copy to components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/PropertyValueMessageGroupIdStrategy.java
index 44caf36..5a51ff7 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/ConstantMessageGroupIdStrategy.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/PropertyValueMessageGroupIdStrategy.java
@@ -18,11 +18,11 @@ package org.apache.camel.component.aws.sqs;
import org.apache.camel.Exchange;
-public class ConstantMessageGroupIdStrategy implements StringValueFromExchangeStrategy {
+public class PropertyValueMessageGroupIdStrategy implements MessageGroupIdStrategy {
@Override
- public String value(Exchange exchange) {
- return "CamelSingleMessageGroup";
+ public String getMessageGroupId(Exchange exchange) {
+ return exchange.getProperty(SqsConstants.MESSAGE_GROUP_ID_PROPERTY, String.class);
}
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
index 653b956..75fabdc 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsComponent.java
@@ -21,6 +21,7 @@ import java.util.Map;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.UriEndpointComponent;
+import org.apache.camel.util.ObjectHelper;
public class SqsComponent extends UriEndpointComponent {
@@ -62,7 +63,7 @@ public class SqsComponent extends UriEndpointComponent {
throw new IllegalArgumentException("Extending message visibility (extendMessageVisibility) requires visibilityTimeout to be set on the Endpoint.");
}
- if (configuration.isFifoQueue() && configuration.getMessageGroupIdStrategy() == null) {
+ if (configuration.isFifoQueue() && ObjectHelper.isEmpty(configuration.getMessageGroupIdStrategy())) {
throw new IllegalArgumentException("messageGroupIdStrategy must be set for FIFO queues.");
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
index dbab657f..24816e3 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
@@ -68,9 +68,9 @@ public class SqsConfiguration {
@UriParam(label = "producer")
private Integer delaySeconds;
@UriParam(label = "producer")
- private StringValueFromExchangeStrategy messageGroupIdStrategy;
+ private MessageGroupIdStrategy messageGroupIdStrategy;
@UriParam(label = "producer", defaultValue = "useExchangeId")
- private StringValueFromExchangeStrategy messageDeduplicationIdStrategy = new ExchangeIdStrategy();
+ private MessageDeduplicationIdStrategy messageDeduplicationIdStrategy = new ExchangeIdMessageDeduplicationIdStrategy();
// queue properties
@UriParam(label = "queue")
@@ -89,7 +89,7 @@ public class SqsConfiguration {
/**
* Whether or not the queue is a FIFO queue
*/
- public boolean isFifoQueue() {
+ boolean isFifoQueue() {
// AWS docs suggest this is valid derivation.
// FIFO queue names must end with .fifo, and standard queues cannot
if (queueName.endsWith(".fifo")) {
@@ -383,26 +383,26 @@ public class SqsConfiguration {
/**
* Since *Camel 2.20*. Only for FIFO queues. Strategy for setting the messageGroupId on the message.
- * Can be one of the following options: *useConstant*, *useExchangeId*, *useHeaderValue*.
- * For the *useHeaderValue* option, the value of header "CamelAwsMessageGroupId" will be used.
+ * Can be one of the following options: *useConstant*, *useExchangeId*, *usePropertyValue*.
+ * For the *usePropertyValue* option, the value of property "CamelAwsMessageGroupId" will be used.
*/
public void setMessageGroupIdStrategy(String strategy) {
if ("useConstant".equalsIgnoreCase(strategy)) {
messageGroupIdStrategy = new ConstantMessageGroupIdStrategy();
} else if ("useExchangeId".equalsIgnoreCase(strategy)) {
- messageGroupIdStrategy = new ExchangeIdStrategy();
- } else if ("useHeaderValue".equalsIgnoreCase(strategy)) {
- messageGroupIdStrategy = new HeaderValueStrategy();
+ messageGroupIdStrategy = new ExchangeIdMessageGroupIdStrategy();
+ } else if ("usePropertyValue".equalsIgnoreCase(strategy)) {
+ messageGroupIdStrategy = new PropertyValueMessageGroupIdStrategy();
} else {
throw new IllegalArgumentException("Unrecognised MessageGroupIdStrategy: " + strategy);
}
}
- public StringValueFromExchangeStrategy getMessageGroupIdStrategy() {
+ public MessageGroupIdStrategy getMessageGroupIdStrategy() {
return messageGroupIdStrategy;
}
- public StringValueFromExchangeStrategy getMessageDeduplicationIdStrategy() {
+ public MessageDeduplicationIdStrategy getMessageDeduplicationIdStrategy() {
return messageDeduplicationIdStrategy;
}
@@ -413,9 +413,9 @@ public class SqsConfiguration {
*/
public void setMessageDeduplicationIdStrategy(String strategy) {
if ("useExchangeId".equalsIgnoreCase(strategy)) {
- messageDeduplicationIdStrategy = new ExchangeIdStrategy();
+ messageDeduplicationIdStrategy = new ExchangeIdMessageDeduplicationIdStrategy();
} else if ("useContentBasedDeduplication".equalsIgnoreCase(strategy)) {
- messageDeduplicationIdStrategy = new NullStrategy();
+ messageDeduplicationIdStrategy = new NullMessageDeduplicationIdStrategy();
} else {
throw new IllegalArgumentException("Unrecognised MessageDeduplicationIdStrategy: " + strategy);
}
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
index 5fc53c6..3c2dec1 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConstants.java
@@ -28,5 +28,5 @@ public interface SqsConstants {
String MESSAGE_ID = "CamelAwsSqsMessageId";
String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
- String MESSAGE_GROUP_ID_HEADER = "CamelAwsMessageGroupId";
+ String MESSAGE_GROUP_ID_PROPERTY = "CamelAwsMessageGroupId";
}
\ No newline at end of file
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
index e3d2f00..dfc2279 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
@@ -72,12 +72,12 @@ public class SqsProducer extends DefaultProducer {
private void configureFifoAttributes(SendMessageRequest request, Exchange exchange) {
if (getEndpoint().getConfiguration().isFifoQueue()) {
// use strategies
- StringValueFromExchangeStrategy messageGroupIdStrategy = getEndpoint().getConfiguration().getMessageGroupIdStrategy();
- String messageGroupId = messageGroupIdStrategy.value(exchange);
+ MessageGroupIdStrategy messageGroupIdStrategy = getEndpoint().getConfiguration().getMessageGroupIdStrategy();
+ String messageGroupId = messageGroupIdStrategy.getMessageGroupId(exchange);
request.setMessageGroupId(messageGroupId);
- StringValueFromExchangeStrategy messageDeduplicationIdStrategy = getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy();
- String messageDeduplicationId = messageDeduplicationIdStrategy.value(exchange);
+ MessageDeduplicationIdStrategy messageDeduplicationIdStrategy = getEndpoint().getConfiguration().getMessageDeduplicationIdStrategy();
+ String messageDeduplicationId = messageDeduplicationIdStrategy.getMessageDeduplicationId(exchange);
request.setMessageDeduplicationId(messageDeduplicationId);
}
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
index c1d24f1..e1e511d 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
@@ -229,8 +229,8 @@ public class SqsProducerTest {
@Test
public void itSetsMessageGroupIdUsingHeaderValueStrategy() throws Exception {
sqsConfiguration.setQueueName("queueName.fifo");
- sqsConfiguration.setMessageGroupIdStrategy("useHeaderValue");
- when(inMessage.getHeader(SqsConstants.MESSAGE_GROUP_ID_HEADER, String.class)).thenReturn("my-group-id");
+ sqsConfiguration.setMessageGroupIdStrategy("usePropertyValue");
+ when(exchange.getProperty(SqsConstants.MESSAGE_GROUP_ID_PROPERTY, String.class)).thenReturn("my-group-id");
underTest.process(exchange);
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 03/03: CAMEL-11224: Polished. This closes #2018
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 52f5e7f1113411b0c6b3b7ab6280867a4745406b
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Oct 27 09:39:14 2017 +0200
CAMEL-11224: Polished. This closes #2018
---
.../camel-aws/src/main/docs/aws-sqs-component.adoc | 4 +++-
.../camel/component/aws/sqs/SqsConfiguration.java | 8 ++++----
.../component/aws/sqs/SqsConfigurationTest.java | 21 +++++++++++++++++++--
3 files changed, 26 insertions(+), 7 deletions(-)
diff --git a/components/camel-aws/src/main/docs/aws-sqs-component.adoc b/components/camel-aws/src/main/docs/aws-sqs-component.adoc
index f619ce4..03dc78c 100644
--- a/components/camel-aws/src/main/docs/aws-sqs-component.adoc
+++ b/components/camel-aws/src/main/docs/aws-sqs-component.adoc
@@ -53,7 +53,7 @@ with the following path and query parameters:
| *queueNameOrArn* | *Required* Queue name or ARN | | String
|===
-==== Query Parameters (45 parameters):
+==== Query Parameters (47 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -80,6 +80,8 @@ with the following path and query parameters:
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
| *pollStrategy* (consumer) | A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange have been created and being routed in Camel. | | PollingConsumerPoll Strategy
| *delaySeconds* (producer) | Delay sending messages for a number of seconds. | | Integer
+| *messageDeduplicationId Strategy* (producer) | Only for FIFO queues. Strategy for setting the messageDeduplicationId on the message. Can be one of the following options: useExchangeId useContentBasedDeduplication. For the useContentBasedDeduplication option no messageDeduplicationId will be set on the message. | useExchangeId | MessageDeduplicationId Strategy
+| *messageGroupIdStrategy* (producer) | Only for FIFO queues. Strategy for setting the messageGroupId on the message. Can be one of the following options: useConstant useExchangeId usePropertyValue. For the usePropertyValue option the value of property CamelAwsMessageGroupId will be used. | | MessageGroupIdStrategy
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
| *backoffErrorThreshold* (scheduler) | The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in. | | int
| *backoffIdleThreshold* (scheduler) | The number of subsequent idle polls that should happen before the backoffMultipler should kick-in. | | int
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
index 24816e3..ec0297d 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsConfiguration.java
@@ -67,9 +67,9 @@ public class SqsConfiguration {
// producer properties
@UriParam(label = "producer")
private Integer delaySeconds;
- @UriParam(label = "producer")
+ @UriParam(label = "producer", enums = "useConstant,useExchangeId,usePropertyValue")
private MessageGroupIdStrategy messageGroupIdStrategy;
- @UriParam(label = "producer", defaultValue = "useExchangeId")
+ @UriParam(label = "producer", defaultValue = "useExchangeId", enums = "useExchangeId,useContentBasedDeduplication")
private MessageDeduplicationIdStrategy messageDeduplicationIdStrategy = new ExchangeIdMessageDeduplicationIdStrategy();
// queue properties
@@ -382,7 +382,7 @@ public class SqsConfiguration {
}
/**
- * Since *Camel 2.20*. Only for FIFO queues. Strategy for setting the messageGroupId on the message.
+ * Only for FIFO queues. Strategy for setting the messageGroupId on the message.
* Can be one of the following options: *useConstant*, *useExchangeId*, *usePropertyValue*.
* For the *usePropertyValue* option, the value of property "CamelAwsMessageGroupId" will be used.
*/
@@ -407,7 +407,7 @@ public class SqsConfiguration {
}
/**
- * Since *Camel 2.20*. Only for FIFO queues. Strategy for setting the messageDeduplicationId on the message.
+ * Only for FIFO queues. Strategy for setting the messageDeduplicationId on the message.
* Can be one of the following options: *useExchangeId*, *useContentBasedDeduplication*.
* For the *useContentBasedDeduplication* option, no messageDeduplicationId will be set on the message.
*/
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
index ace5173..3c7874a 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsConfigurationTest.java
@@ -1,9 +1,26 @@
+/**
+ * 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.aws.sqs;
-import static org.junit.Assert.*;
-
import org.junit.Test;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
public class SqsConfigurationTest {
@Test
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.