receive sqs message attributes using the camel dsl?

0
votes

Anybody knows how to receive SQS message attributes using the Camel DSL in java? I'm getting the following error:

"Failed to create route payee route: Route(batch route)[[From[aws-sqs://myqueue?amazonSQSEndpoint=... because of Failed to resolve endpoint: aws-sqs://myqueue?amazonSQSEndpoint=sqs.us-west-1.amazonaws.com&accessKey=*****&secretKey=****************&maxMessagesPerPoll=1&messageAttributeNames=%5BuserID%5 due to: Could not find a suitable setter for property: messageAttributeNames as there isn't a setter method with same type: java.lang.String nor type conversion possible: No type converter available to convert from type: java.lang.String to the required type: java.util.Collection with value [userID] "

Please find my code 

StringBuilder QueueURI = new StringBuilder();
QueueURI(PropertyUtils.AWS_SQS)
        .append(propertyUtils.queueName)
        .append(PropertyUtils.AMAZON_SQS_REGION)
        .append(propertyUtils.sqsRegion);
QueueURI(PropertyUtils.AWS_ACCESS_KEY).append(
         propertyUtils.awsAccessKey);
QueueURI(PropertyUtils.AWS_SECRET_KEY).append(
         propertyUtils.awsSecretKey);
QueueURI(PropertyUtils.MAX_MESSAGES_PER_POLL_1);
QueueURI("&messageAttributeNames=");


Collection<String> collection = new ArrayList<String>();
collection.add("userID");

//aws-sqs://myqueue?amazonSQSEndpoint=sqs.us-west-1.amazonaws.com&accessKey=*****&secretKey=****************&maxMessagesPerPoll=1&messageAttributeNames=[userID]

from(QueueURI.ToString() + collection)
       .routeId("batch route")
       .process(userValidator);
2
javaamazon-web-servicesapache-camelamazon-sqs

2 Answers

1
votes

Camel does not have a java.lang.String => java.util.Collection TypeConverter by default. You can implement a org.apache.camel.TypeConverter, which can then be registered with the CamelContext's TypeConverterRegistry.

I am using Spring, so I leveraged Spring's conversion support:

import org.apache.camel.Exchange;
import org.apache.camel.TypeConversionException;
import org.apache.camel.support.TypeConverterSupport;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;

public class TypeConverterBridge extends TypeConverterSupport {
    private ConversionService cs = new DefaultConversionService();

    @Override
    public <T> T convertTo(Class<T> type, Exchange exchange, Object value) throws TypeConversionException {
        if (cs.canConvert(value.getClass(), type)) {
            return cs.convert(value, type);
        }
        return null;
    }
}

And then registered the TypeConverter with my CamelContext:

camelContext.getTypeConverterRegistry().addFallbackTypeConverter(new TypeConverterBridge(), false);
0
votes

You can find the attributes of your SQS messages in a header called CamelAwsSqsAttributes as explained here: http://camel.apache.org/aws-sqs.html

That header is a Map<String, String> that holds what you are looking for. If you want to see them, you can do something like:

...
from(QueueURI.ToString() + collection)
   .routeId("batch route")
   .log("Attributes: ${header.CamelAwsSqsAttributes}")
   .process(userValidator);