You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2017/12/07 14:08:15 UTC
[camel] 01/03: CAMEL-12071 aws-sqs queue creation does not support
FIFO queues
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 33cc329ebd58bdb569357fde90ff8907d8062e48
Author: PyvesB <pi...@bbc.co.uk>
AuthorDate: Thu Dec 7 11:41:54 2017 +0000
CAMEL-12071 aws-sqs queue creation does not support FIFO queues
---
.../camel/component/aws/sqs/SqsEndpoint.java | 30 +++++++------
.../camel/component/aws/sqs/SqsEndpointTest.java | 51 +++++++++++++++++++---
2 files changed, 63 insertions(+), 18 deletions(-)
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsEndpoint.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsEndpoint.java
index 8ba00c4..f1b2dc3 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsEndpoint.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsEndpoint.java
@@ -26,7 +26,6 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
import com.amazonaws.services.sqs.AmazonSQS;
-import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.CreateQueueResult;
@@ -62,7 +61,7 @@ import org.slf4j.LoggerFactory;
@UriEndpoint(firstVersion = "2.6.0", scheme = "aws-sqs", title = "AWS Simple Queue Service", syntax = "aws-sqs:queueNameOrArn",
consumerClass = SqsConsumer.class, label = "cloud,messaging")
public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterStrategyAware {
-
+
private static final Logger LOG = LoggerFactory.getLogger(SqsEndpoint.class);
private AmazonSQS client;
@@ -82,7 +81,7 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
super(uri, component);
this.configuration = configuration;
}
-
+
public HeaderFilterStrategy getHeaderFilterStrategy() {
return headerFilterStrategy;
}
@@ -93,7 +92,7 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
this.headerFilterStrategy = strategy;
}
-
+
public Producer createProducer() throws Exception {
return new SqsProducer(this);
}
@@ -116,12 +115,12 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
protected void doStart() throws Exception {
client = getConfiguration().getAmazonSQSClient() != null
? getConfiguration().getAmazonSQSClient() : getClient();
-
+
// Override the endpoint location
if (ObjectHelper.isNotEmpty(getConfiguration().getAmazonSQSEndpoint())) {
client.setEndpoint(getConfiguration().getAmazonSQSEndpoint());
}
-
+
// check the setting the headerFilterStrategy
if (headerFilterStrategy == null) {
headerFilterStrategy = new SqsHeaderFilterStrategy();
@@ -165,6 +164,11 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
// creates a new queue, or returns the URL of an existing one
CreateQueueRequest request = new CreateQueueRequest(configuration.getQueueName());
+ if (getConfiguration().isFifoQueue()) {
+ request.getAttributes().put(QueueAttributeName.FifoQueue.name(), String.valueOf(true));
+ boolean useContentBasedDeduplication = getConfiguration().getMessageDeduplicationIdStrategy() instanceof NullMessageDeduplicationIdStrategy;
+ request.getAttributes().put(QueueAttributeName.ContentBasedDeduplication.name(), String.valueOf(useContentBasedDeduplication));
+ }
if (getConfiguration().getDefaultVisibilityTimeout() != null) {
request.getAttributes().put(QueueAttributeName.VisibilityTimeout.name(), String.valueOf(getConfiguration().getDefaultVisibilityTimeout()));
}
@@ -184,10 +188,10 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
request.getAttributes().put(QueueAttributeName.RedrivePolicy.name(), getConfiguration().getRedrivePolicy());
}
LOG.trace("Creating queue [{}] with request [{}]...", configuration.getQueueName(), request);
-
+
CreateQueueResult queueResult = client.createQueue(request);
queueUrl = queueResult.getQueueUrl();
-
+
LOG.trace("Queue created and available at: {}", queueUrl);
}
@@ -238,10 +242,10 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
message.setHeader(SqsConstants.RECEIPT_HANDLE, msg.getReceiptHandle());
message.setHeader(SqsConstants.ATTRIBUTES, msg.getAttributes());
message.setHeader(SqsConstants.MESSAGE_ATTRIBUTES, msg.getMessageAttributes());
-
+
//Need to apply the SqsHeaderFilterStrategy this time
HeaderFilterStrategy headerFilterStrategy = getHeaderFilterStrategy();
- //add all sqs message attributes as camel message headers so that knowledge of
+ //add all sqs message attributes as camel message headers so that knowledge of
//the Sqs class MessageAttributeValue will not leak to the client
for (Entry<String, MessageAttributeValue> entry : msg.getMessageAttributes().entrySet()) {
String header = entry.getKey();
@@ -260,14 +264,14 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
public void setConfiguration(SqsConfiguration configuration) {
this.configuration = configuration;
}
-
+
public AmazonSQS getClient() {
if (client == null) {
client = createClient();
}
return client;
}
-
+
public void setClient(AmazonSQS client) {
this.client = client;
}
@@ -326,7 +330,7 @@ public class SqsEndpoint extends ScheduledPollEndpoint implements HeaderFilterSt
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
-
+
private Object translateValue(MessageAttributeValue mav) {
Object result = null;
if (mav.getStringValue() != null) {
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsEndpointTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsEndpointTest.java
index 56a5fcb..d7122e1 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsEndpointTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsEndpointTest.java
@@ -17,31 +17,36 @@
package org.apache.camel.component.aws.sqs;
import com.amazonaws.services.sqs.AmazonSQSClient;
+import com.amazonaws.services.sqs.model.CreateQueueRequest;
+import com.amazonaws.services.sqs.model.CreateQueueResult;
import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
import com.amazonaws.services.sqs.model.GetQueueUrlResult;
import com.amazonaws.services.sqs.model.ListQueuesResult;
+import com.amazonaws.services.sqs.model.QueueAttributeName;
import org.apache.camel.impl.DefaultCamelContext;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
+import static org.junit.Assert.assertEquals;
public class SqsEndpointTest {
-
+
private SqsEndpoint endpoint;
private AmazonSQSClient amazonSQSClient;
+ private SqsConfiguration config;
@Before
public void setUp() throws Exception {
amazonSQSClient = Mockito.mock(AmazonSQSClient.class);
-
- SqsConfiguration config = new SqsConfiguration();
+
+ config = new SqsConfiguration();
config.setQueueName("test-queue");
config.setAmazonSQSClient(amazonSQSClient);
-
+
endpoint = new SqsEndpoint("aws-sqs://test-queue", new SqsComponent(new DefaultCamelContext()), config);
-
+
}
@Test
@@ -69,4 +74,40 @@ public class SqsEndpointTest {
Mockito.verify(amazonSQSClient).getQueueUrl(expectedGetQueueUrlRequest);
}
+
+ @Test
+ public void createQueueShouldCreateFifoQueueWithContentBasedDeduplication() {
+ config.setQueueName("test-queue.fifo");
+ config.setMessageDeduplicationIdStrategy("useContentBasedDeduplication");
+
+ CreateQueueRequest expectedCreateQueueRequest = new CreateQueueRequest("test-queue.fifo")
+ .addAttributesEntry(QueueAttributeName.FifoQueue.name(), "true")
+ .addAttributesEntry(QueueAttributeName.ContentBasedDeduplication.name(), "true");
+ Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ .thenReturn(new CreateQueueResult()
+ .withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo"));
+
+ endpoint.createQueue(amazonSQSClient);
+
+ Mockito.verify(amazonSQSClient).createQueue(expectedCreateQueueRequest);
+ assertEquals("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo", endpoint.getQueueUrl());
+ }
+
+ @Test
+ public void createQueueShouldCreateFifoQueueWithoutContentBasedDeduplication() {
+ config.setQueueName("test-queue.fifo");
+ config.setMessageDeduplicationIdStrategy("useExchangeId");
+
+ CreateQueueRequest expectedCreateQueueRequest = new CreateQueueRequest("test-queue.fifo")
+ .addAttributesEntry(QueueAttributeName.FifoQueue.name(), "true")
+ .addAttributesEntry(QueueAttributeName.ContentBasedDeduplication.name(), "false");
+ Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ .thenReturn(new CreateQueueResult()
+ .withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo"));
+
+ endpoint.createQueue(amazonSQSClient);
+
+ Mockito.verify(amazonSQSClient).createQueue(expectedCreateQueueRequest);
+ assertEquals("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo", endpoint.getQueueUrl());
+ }
}
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.