camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ningji...@apache.org
Subject [2/2] git commit: CAMEL-7654 Support Message attribute for the SQSProducer
Date Wed, 13 Aug 2014 09:50:40 GMT
CAMEL-7654 Support Message attribute for the SQSProducer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/2ca8187c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/2ca8187c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/2ca8187c

Branch: refs/heads/master
Commit: 2ca8187cf079b011b7f4438d803cd51c96378dc9
Parents: e7c9e40
Author: Willem Jiang <willem.jiang@gmail.com>
Authored: Wed Aug 13 16:58:21 2014 +0800
Committer: Willem Jiang <willem.jiang@gmail.com>
Committed: Wed Aug 13 17:49:50 2014 +0800

----------------------------------------------------------------------
 .../camel/component/aws/sqs/SqsEndpoint.java    | 18 +++++
 .../camel/component/aws/sqs/SqsProducer.java    | 24 +++++++
 .../component/aws/sqs/SqsProducerTest.java      | 72 ++++++++++++++++++++
 3 files changed, 114 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/2ca8187c/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsEndpoint.java
----------------------------------------------------------------------
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 6447d7b..b48fbe6 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
@@ -17,6 +17,7 @@
 package org.apache.camel.component.aws.sqs;
 
 import java.util.HashMap;
+import java.util.Map.Entry;
 
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.BasicAWSCredentials;
@@ -27,6 +28,7 @@ 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.MessageAttributeValue;
 import com.amazonaws.services.sqs.model.QueueAttributeName;
 import com.amazonaws.services.sqs.model.SetQueueAttributesRequest;
 
@@ -188,6 +190,12 @@ public class SqsEndpoint extends ScheduledPollEndpoint {
         message.setHeader(SqsConstants.RECEIPT_HANDLE, msg.getReceiptHandle());
         message.setHeader(SqsConstants.ATTRIBUTES, msg.getAttributes());
         message.setHeader(SqsConstants.MESSAGE_ATTRIBUTES, msg.getMessageAttributes());
+        
+        //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())
{
+            message.setHeader(entry.getKey(), translateValue(entry.getValue()));
+        }
         return exchange;
     }
 
@@ -231,4 +239,14 @@ public class SqsEndpoint extends ScheduledPollEndpoint {
     public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
         this.maxMessagesPerPoll = maxMessagesPerPoll;
     }
+    
+    private Object translateValue(MessageAttributeValue mav) {
+        Object result = null;
+        if (mav.getStringValue() != null) {
+            result = mav.getStringValue();
+        } else if (mav.getBinaryValue() != null) {
+            result = mav.getBinaryValue();
+        }
+        return result;
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/2ca8187c/components/camel-aws/src/main/java/org/apache/camel/component/aws/sqs/SqsProducer.java
----------------------------------------------------------------------
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 b4fa72b..0a4c948 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
@@ -16,7 +16,13 @@
  */
 package org.apache.camel.component.aws.sqs;
 
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
 import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
 import com.amazonaws.services.sqs.model.SendMessageRequest;
 import com.amazonaws.services.sqs.model.SendMessageResult;
 
@@ -44,6 +50,7 @@ public class SqsProducer extends DefaultProducer {
     public void process(Exchange exchange) throws Exception {
         String body = exchange.getIn().getBody(String.class);
         SendMessageRequest request = new SendMessageRequest(getQueueUrl(), body);
+        request.setMessageAttributes(translateAttributes(exchange.getIn().getHeaders()));
         addDelay(request, exchange);
 
         LOG.trace("Sending request [{}] from exchange [{}]...", request, exchange);
@@ -98,4 +105,21 @@ public class SqsProducer extends DefaultProducer {
     public String toString() {
         return "SqsProducer[" + URISupport.sanitizeUri(getEndpoint().getEndpointUri()) +
"]";
     }
+    
+    private Map<String, MessageAttributeValue> translateAttributes(Map<String, Object>
headers) {
+        Map<String, MessageAttributeValue> result = new HashMap<String, MessageAttributeValue>();
+        for (Entry<String, Object> entry : headers.entrySet()) {
+            Object value = entry.getValue();
+            MessageAttributeValue mav = new MessageAttributeValue();
+            if (value instanceof String) {
+                mav.setDataType("String");
+                mav.withStringValue((String)value);
+            } else if (value instanceof ByteBuffer) {
+                mav.setDataType("Binary");
+                mav.withBinaryValue((ByteBuffer)value);
+            }
+            result.put(entry.getKey(), mav);
+        }
+        return result;
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/2ca8187c/components/camel-aws/src/test/java/org/apache/camel/component/aws/sqs/SqsProducerTest.java
----------------------------------------------------------------------
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 efc4f0e..e6742ca 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
@@ -16,6 +16,10 @@
  */
 package org.apache.camel.component.aws.sqs;
 
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
 import com.amazonaws.services.sqs.AmazonSQSClient;
 import com.amazonaws.services.sqs.model.SendMessageRequest;
 import com.amazonaws.services.sqs.model.SendMessageResult;
@@ -31,6 +35,7 @@ import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.mock;
@@ -43,6 +48,12 @@ public class SqsProducerTest {
     private static final String MESSAGE_MD5 = "00000000000000000000000000000000";
     private static final String MESSAGE_ID = "11111111111111111111111111111111";
     private static final String QUEUE_URL = "some://queue/url";
+    private static final String SAMPLE_MESSAGE_HEADER_NAME_1 = "header_name_1";
+    private static final String SAMPLE_MESSAGE_HEADER_VALUE_1 = "heder_value_1";
+    private static final String SAMPLE_MESSAGE_HEADER_NAME_2 = "header_name_2";
+    private static final ByteBuffer SAMPLE_MESSAGE_HEADER_VALUE_2 = ByteBuffer.wrap(new byte[10]);
+    private static final String SAMPLE_MESSAGE_HEADER_NAME_3 = "header_name_3";
+    private static final String SAMPLE_MESSAGE_HEADER_VALUE_3 = "heder_value_3";
     
     Exchange exchange = mock(Exchange.class, RETURNS_DEEP_STUBS);
 
@@ -121,5 +132,66 @@ public class SqsProducerTest {
         underTest.process(exchange);
         verify(inMessage).setHeader(SqsConstants.MD5_OF_BODY, MESSAGE_MD5);
     }
+    
+    @Test
+    public void isAttributeMessageStringHeaderOnTheRequest() throws Exception {
+        Map<String, Object> headers = new HashMap<String, Object>();
+        headers.put(SAMPLE_MESSAGE_HEADER_NAME_1, SAMPLE_MESSAGE_HEADER_VALUE_1);
+        when(inMessage.getHeaders()).thenReturn(headers);
+        underTest.process(exchange);
+
+        ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+        verify(amazonSQSClient).sendMessage(capture.capture());
+
+        assertEquals(SAMPLE_MESSAGE_HEADER_VALUE_1,
+                     capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_1)
+                         .getStringValue());
+        assertNull(capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_1)
+            .getBinaryValue());
+    }
+    
+    @Test
+    public void isAttributeMessageByteBufferHeaderOnTheRequest() throws Exception {
+        Map<String, Object> headers = new HashMap<String, Object>();
+        headers.put(SAMPLE_MESSAGE_HEADER_NAME_2, SAMPLE_MESSAGE_HEADER_VALUE_2);
+        when(inMessage.getHeaders()).thenReturn(headers);
+        underTest.process(exchange);
+
+        ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+        verify(amazonSQSClient).sendMessage(capture.capture());
+
+        assertEquals(SAMPLE_MESSAGE_HEADER_VALUE_2,
+                     capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_2)
+                         .getBinaryValue());
+        assertNull(capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_2)
+            .getStringValue());
+    }
+
+    @Test
+    public void isAllAttributeMessagesOnTheRequest() throws Exception {
+        Map<String, Object> headers = new HashMap<String, Object>();
+        headers.put(SAMPLE_MESSAGE_HEADER_NAME_1, SAMPLE_MESSAGE_HEADER_VALUE_1);
+        headers.put(SAMPLE_MESSAGE_HEADER_NAME_2, SAMPLE_MESSAGE_HEADER_VALUE_2);
+        headers.put(SAMPLE_MESSAGE_HEADER_NAME_3, SAMPLE_MESSAGE_HEADER_VALUE_3);
+        when(inMessage.getHeaders()).thenReturn(headers);
+        underTest.process(exchange);
+
+        ArgumentCaptor<SendMessageRequest> capture = ArgumentCaptor.forClass(SendMessageRequest.class);
+        verify(amazonSQSClient).sendMessage(capture.capture());
+
+        assertEquals(SAMPLE_MESSAGE_HEADER_VALUE_1,
+                     capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_1)
+                         .getStringValue());
+        assertEquals(SAMPLE_MESSAGE_HEADER_VALUE_2,
+                     capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_2)
+                         .getBinaryValue());
+        assertEquals(SAMPLE_MESSAGE_HEADER_VALUE_3,
+                     capture.getValue().getMessageAttributes().get(SAMPLE_MESSAGE_HEADER_NAME_3)
+                         .getStringValue());
+        assertEquals(3, capture.getValue().getMessageAttributes().size());
+    }
+    
+    
+
 
 }
\ No newline at end of file


Mime
View raw message