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:14 UTC
[camel] branch master updated (0a9790b -> 45fd014)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 0a9790b Camel-Consul fixed Package name
new 33cc329 CAMEL-12071 aws-sqs queue creation does not support FIFO queues
new e872461 CAMEL-12071 improved unit tests for createQueue
new 45fd014 CAMEL-12071 made unit tests fail in expected way
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/component/aws/sqs/SqsEndpoint.java | 30 +++++----
.../camel/component/aws/sqs/SqsEndpointTest.java | 77 ++++++++++++++++++++--
2 files changed, 89 insertions(+), 18 deletions(-)
--
To stop receiving notification emails like this one, please contact
['"commits@camel.apache.org" <co...@camel.apache.org>'].
[camel] 01/03: CAMEL-12071 aws-sqs queue creation does not support
FIFO queues
Posted by ac...@apache.org.
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>.
[camel] 02/03: CAMEL-12071 improved unit tests for createQueue
Posted by ac...@apache.org.
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 e8724613dcc8c498263379ba113bf5477c44f4cc
Author: PyvesB <pi...@bbc.co.uk>
AuthorDate: Thu Dec 7 11:56:57 2017 +0000
CAMEL-12071 improved unit tests for createQueue
---
.../camel/component/aws/sqs/SqsEndpointTest.java | 26 ++++++++++++++++++++++
1 file changed, 26 insertions(+)
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 d7122e1..18aef1e 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
@@ -110,4 +110,30 @@ public class SqsEndpointTest {
Mockito.verify(amazonSQSClient).createQueue(expectedCreateQueueRequest);
assertEquals("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo", endpoint.getQueueUrl());
}
+
+ @Test
+ public void createQueueShouldCreateStandardQueueWithCorrectAttributes() {
+ config.setDefaultVisibilityTimeout(1000);
+ config.setMaximumMessageSize(128);
+ config.setMessageRetentionPeriod(1000);
+ config.setPolicy("{\"Version\": \"2012-10-17\"}");
+ config.setReceiveMessageWaitTimeSeconds(5);
+ config.setRedrivePolicy("{ \"deadLetterTargetArn\" : String, \"maxReceiveCount\" : Integer }");
+
+ CreateQueueRequest expectedCreateQueueRequest = new CreateQueueRequest("test-queue")
+ .addAttributesEntry(QueueAttributeName.VisibilityTimeout.name(), "1000")
+ .addAttributesEntry(QueueAttributeName.MaximumMessageSize.name(), "128")
+ .addAttributesEntry(QueueAttributeName.MessageRetentionPeriod.name(), "1000")
+ .addAttributesEntry(QueueAttributeName.Policy.name(), "{\"Version\": \"2012-10-17\"}")
+ .addAttributesEntry(QueueAttributeName.ReceiveMessageWaitTimeSeconds.name(), "5")
+ .addAttributesEntry(QueueAttributeName.RedrivePolicy.name(), "{ \"deadLetterTargetArn\" : String, \"maxReceiveCount\" : Integer }");
+ Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ .thenReturn(new CreateQueueResult()
+ .withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue"));
+
+ endpoint.createQueue(amazonSQSClient);
+
+ Mockito.verify(amazonSQSClient).createQueue(expectedCreateQueueRequest);
+ assertEquals("https://sqs.us-east-1.amazonaws.com/111222333/test-queue", 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>.
[camel] 03/03: CAMEL-12071 made unit tests fail in expected way
Posted by ac...@apache.org.
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 45fd0143a461122f63f54ba1f9af5a6334d9afa1
Author: PyvesB <pi...@bbc.co.uk>
AuthorDate: Thu Dec 7 13:59:48 2017 +0000
CAMEL-12071 made unit tests fail in expected way
---
.../java/org/apache/camel/component/aws/sqs/SqsEndpointTest.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
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 18aef1e..4f4eb22 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
@@ -83,7 +83,7 @@ public class SqsEndpointTest {
CreateQueueRequest expectedCreateQueueRequest = new CreateQueueRequest("test-queue.fifo")
.addAttributesEntry(QueueAttributeName.FifoQueue.name(), "true")
.addAttributesEntry(QueueAttributeName.ContentBasedDeduplication.name(), "true");
- Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ Mockito.when(amazonSQSClient.createQueue(Mockito.any(CreateQueueRequest.class)))
.thenReturn(new CreateQueueResult()
.withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo"));
@@ -101,7 +101,7 @@ public class SqsEndpointTest {
CreateQueueRequest expectedCreateQueueRequest = new CreateQueueRequest("test-queue.fifo")
.addAttributesEntry(QueueAttributeName.FifoQueue.name(), "true")
.addAttributesEntry(QueueAttributeName.ContentBasedDeduplication.name(), "false");
- Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ Mockito.when(amazonSQSClient.createQueue(Mockito.any(CreateQueueRequest.class)))
.thenReturn(new CreateQueueResult()
.withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue.fifo"));
@@ -127,7 +127,7 @@ public class SqsEndpointTest {
.addAttributesEntry(QueueAttributeName.Policy.name(), "{\"Version\": \"2012-10-17\"}")
.addAttributesEntry(QueueAttributeName.ReceiveMessageWaitTimeSeconds.name(), "5")
.addAttributesEntry(QueueAttributeName.RedrivePolicy.name(), "{ \"deadLetterTargetArn\" : String, \"maxReceiveCount\" : Integer }");
- Mockito.when(amazonSQSClient.createQueue(expectedCreateQueueRequest))
+ Mockito.when(amazonSQSClient.createQueue(Mockito.any(CreateQueueRequest.class)))
.thenReturn(new CreateQueueResult()
.withQueueUrl("https://sqs.us-east-1.amazonaws.com/111222333/test-queue"));
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.