This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 46b3a1949ab5 CAMEL-26300: camel-aws2-sqs, camel-aws2-ddb - the SQS
CloudEvent source names the queue, and the DynamoDB x-struct transformer reads
stream and file bodies as text (#29216)
46b3a1949ab5 is described below
commit 46b3a1949ab5490ae8a99fd30f51d36f3ba325bb
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 13:03:35 2026 +0100
CAMEL-26300: camel-aws2-sqs, camel-aws2-ddb - the SQS CloudEvent source
names the queue, and the DynamoDB x-struct transformer reads stream and file
bodies as text (#29216)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/aws2-sqs.json | 3 +-
.../aws2-ddb-application-x-struct.json | 6 +-
.../transformer/aws2-ddb-application-x-struct.json | 6 +-
.../Ddb2JsonStructDataTypeTransformer.java | 44 ++++++--
.../Ddb2JsonStructDataTypeTransformerTest.java | 90 +++++++++++++---
.../apache/camel/component/aws2/sqs/aws2-sqs.json | 3 +-
.../camel/component/aws2/sqs/Sqs2Constants.java | 4 +
.../camel/component/aws2/sqs/Sqs2Consumer.java | 21 ++++
.../AWS2SQSCloudEventDataTypeTransformer.java | 7 +-
...2SQSCloudEventDataTypeTransformerRouteTest.java | 118 +++++++++++++++++++++
.../AWS2SQSCloudEventDataTypeTransformerTest.java | 16 ++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 20 ++++
.../endpoint/dsl/Sqs2EndpointBuilderFactory.java | 12 +++
13 files changed, 319 insertions(+), 31 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
index ebcc76e86d44..f13e3f148eb7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
@@ -91,7 +91,8 @@
"CamelAwsSqsIsTruncated": { "index": 10, "kind": "header", "displayName":
"", "group": "listQueues", "label": "listQueues", "required": false,
"javaType": "Boolean", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Whether the response has more results
(is truncated).", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#IS_TRUNCATED" },
"CamelAwsSqsSequenceNumber": { "index": 11, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The sequence number for FIFO queues.",
"constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SEQUENCE_NUMBER" },
"CamelAwsSqsFailedMessageCount": { "index": 12, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of failed messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#FAILED_MESSAGE_COUNT" },
- "CamelAwsSqsSuccessfulMessageCount": { "index": 13, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of successful messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SUCCESSFUL_MESSAGE_COUNT" }
+ "CamelAwsSqsSuccessfulMessageCount": { "index": 13, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of successful messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SUCCESSFUL_MESSAGE_COUNT" },
+ "CamelAwsSqsQueueName": { "index": 14, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The name of the queue the message was received
from.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#QUEUE_NAME" }
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String",
"description": "The body of the SQS message" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/transformers/aws2-ddb-application-x-struct.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/transformers/aws2-ddb-application-x-struct.json
index 415e29f1131b..6392afe122e2 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/transformers/aws2-ddb-application-x-struct.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/transformers/aws2-ddb-application-x-struct.json
@@ -3,7 +3,7 @@
"kind": "transformer",
"name": "aws2-ddb:application-x-struct",
"title": "Aws2 Ddb (Application X Struct)",
- "description": "Transforms DynamoDB record into a Json node",
+ "description": "Transforms DynamoDB record into a Json String",
"deprecated": false,
"javaType":
"org.apache.camel.component.aws2.ddb.transform.Ddb2JsonStructDataTypeTransformer",
"groupId": "org.apache.camel",
@@ -15,12 +15,14 @@
"carriers": [
"text",
"bytes",
+ "stream",
+ "file",
"value",
"map",
"list",
"pojo"
],
- "description": "A String is left as it is; another body, such as the
Record of the aws2-ddbstream consumer, is written as JSON with Gson, which
fails on a stream or a file"
+ "description": "A String is left as it is, and bytes, a stream or a file
are read as text; another body, such as the Record of the aws2-ddbstream
consumer, is written as JSON with Gson"
},
"transformerOut": {
"carriers": [
diff --git
a/components/camel-aws/camel-aws2-ddb/src/generated/resources/META-INF/services/org/apache/camel/transformer/aws2-ddb-application-x-struct.json
b/components/camel-aws/camel-aws2-ddb/src/generated/resources/META-INF/services/org/apache/camel/transformer/aws2-ddb-application-x-struct.json
index 415e29f1131b..6392afe122e2 100644
---
a/components/camel-aws/camel-aws2-ddb/src/generated/resources/META-INF/services/org/apache/camel/transformer/aws2-ddb-application-x-struct.json
+++
b/components/camel-aws/camel-aws2-ddb/src/generated/resources/META-INF/services/org/apache/camel/transformer/aws2-ddb-application-x-struct.json
@@ -3,7 +3,7 @@
"kind": "transformer",
"name": "aws2-ddb:application-x-struct",
"title": "Aws2 Ddb (Application X Struct)",
- "description": "Transforms DynamoDB record into a Json node",
+ "description": "Transforms DynamoDB record into a Json String",
"deprecated": false,
"javaType":
"org.apache.camel.component.aws2.ddb.transform.Ddb2JsonStructDataTypeTransformer",
"groupId": "org.apache.camel",
@@ -15,12 +15,14 @@
"carriers": [
"text",
"bytes",
+ "stream",
+ "file",
"value",
"map",
"list",
"pojo"
],
- "description": "A String is left as it is; another body, such as the
Record of the aws2-ddbstream consumer, is written as JSON with Gson, which
fails on a stream or a file"
+ "description": "A String is left as it is, and bytes, a stream or a file
are read as text; another body, such as the Record of the aws2-ddbstream
consumer, is written as JSON with Gson"
},
"transformerOut": {
"carriers": [
diff --git
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformer.java
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformer.java
index a00d85433a15..c06dd8b0b51f 100644
---
a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformer.java
+++
b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformer.java
@@ -17,11 +17,19 @@
package org.apache.camel.component.aws2.ddb.transform;
+import java.io.File;
+import java.io.InputStream;
+import java.io.Reader;
+import java.nio.file.Path;
import java.time.Instant;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
import
org.apache.camel.component.aws2.ddb.transform.serialization.gson.JavaTimeInstantTypeAdapter;
import org.apache.camel.spi.BodyType;
import org.apache.camel.spi.BodyType.Carrier;
@@ -30,13 +38,18 @@ import org.apache.camel.spi.DataType;
import org.apache.camel.spi.DataTypeTransformer;
import org.apache.camel.spi.Transformer;
+/**
+ * Writes the body as a Json String: a String is left as it is, bytes, a
stream or a file are read as text, and another
+ * body, such as the Record of the aws2-ddbstream consumer, is written as Json
with Gson.
+ */
@DataTypeTransformer(name = "aws2-ddb:application-x-struct",
- description = "Transforms DynamoDB record into a Json
node")
+ description = "Transforms DynamoDB record into a Json
String")
@BodyTypes(transformerIn = @BodyType(value = {
- Carrier.TEXT, Carrier.BYTES, Carrier.VALUE, Carrier.MAP, Carrier.LIST,
Carrier.POJO },
- description = "A String is left as it is;
another body, such as the Record of the"
- + " aws2-ddbstream
consumer, is written as JSON with Gson, which fails on a"
- + " stream or a file"),
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.VALUE, Carrier.MAP, Carrier.LIST,
+ Carrier.POJO },
+ description = "A String is left as it is,
and bytes, a stream or a file are read as"
+ + " text; another body,
such as the Record of the aws2-ddbstream"
+ + " consumer, is written as
JSON with Gson"),
transformerOut = @BodyType(value = Carrier.TEXT, javaType =
"java.lang.String"))
public class Ddb2JsonStructDataTypeTransformer extends Transformer {
@@ -46,10 +59,27 @@ public class Ddb2JsonStructDataTypeTransformer extends
Transformer {
@Override
public void transform(Message message, DataType fromType, DataType toType)
{
- if (message.getBody() instanceof String) {
+ Object body = message.getBody();
+ if (body instanceof String) {
+ return;
+ }
+
+ if (body instanceof byte[] || isStreamOrFile(body)) {
+ // their content is the Json, as for a String: Gson would write
bytes as an array of numbers, and cannot
+ // write a stream or a file, whose fields are of JDK classes it
may not reflect into
+ try {
+ message.setBody(message.getMandatoryBody(String.class));
+ } catch (InvalidPayloadException e) {
+ throw new CamelExecutionException("Failed to read the stream
or file body as text", message.getExchange(), e);
+ }
return;
}
- message.setBody(gson.toJson(message.getBody()));
+ message.setBody(gson.toJson(body));
+ }
+
+ private static boolean isStreamOrFile(Object body) {
+ return body instanceof InputStream || body instanceof Reader || body
instanceof StreamCache
+ || body instanceof File || body instanceof Path || body
instanceof WrappedFile;
}
}
diff --git
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformerTest.java
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformerTest.java
index 807321195e05..e5b57bbe6830 100644
---
a/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformerTest.java
+++
b/components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddb/transform/Ddb2JsonStructDataTypeTransformerTest.java
@@ -18,25 +18,32 @@ package org.apache.camel.component.aws2.ddb.transform;
import java.io.ByteArrayInputStream;
import java.io.File;
+import java.io.StringReader;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.time.Instant;
import java.util.List;
import java.util.Map;
-import com.google.gson.JsonIOException;
+import org.apache.camel.CamelExecutionException;
import org.apache.camel.Exchange;
+import org.apache.camel.component.file.FileConsumer;
+import org.apache.camel.converter.stream.InputStreamCache;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.spi.BodyType.Carrier;
import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.DataType;
import org.apache.camel.support.DefaultExchange;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.OperationType;
import software.amazon.awssdk.services.dynamodb.model.Record;
import software.amazon.awssdk.services.dynamodb.model.StreamRecord;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -44,15 +51,21 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
class Ddb2JsonStructDataTypeTransformerTest {
+ private static final String JSON = "{\"name\": \"Rajesh\"}";
+
private final DefaultCamelContext camelContext = new DefaultCamelContext();
private final Ddb2JsonStructDataTypeTransformer transformer = new
Ddb2JsonStructDataTypeTransformer();
+ @TempDir
+ Path dir;
+
@Test
void theBodyTypesSayTheBody() throws Exception {
// CAMEL-25928: the bodies transformerIn says, and a String after, as
transformerOut says
BodyTypes bodyTypes =
Ddb2JsonStructDataTypeTransformer.class.getAnnotation(BodyTypes.class);
- assertEquals(List.of(Carrier.TEXT, Carrier.BYTES, Carrier.VALUE,
Carrier.MAP, Carrier.LIST, Carrier.POJO),
+ assertEquals(List.of(Carrier.TEXT, Carrier.BYTES, Carrier.STREAM,
Carrier.FILE, Carrier.VALUE, Carrier.MAP,
+ Carrier.LIST, Carrier.POJO),
List.of(bodyTypes.transformerIn().value()));
assertEquals(Carrier.TEXT, bodyTypes.transformerOut().value()[0]);
assertEquals(String.class.getCanonicalName(),
bodyTypes.transformerOut().javaType());
@@ -60,30 +73,79 @@ class Ddb2JsonStructDataTypeTransformerTest {
.dynamodb(StreamRecord.builder().approximateCreationDateTime(Instant.ofEpochSecond(1))
.newImage(Map.of("name",
AttributeValue.builder().s("Rajesh").build())).build())
.build();
- for (Object body : List.of(record,
"Test".getBytes(StandardCharsets.UTF_8), 5, Map.of("key", "value"),
- List.of("a", "b"))) {
+ Path file = Files.writeString(dir.resolve("item.json"), JSON);
+ for (Object body : List.of(record,
"Test".getBytes(StandardCharsets.UTF_8),
+ new
ByteArrayInputStream(JSON.getBytes(StandardCharsets.UTF_8)), file.toFile(), 5,
+ Map.of("key", "value"), List.of("a", "b"))) {
Exchange exchange = new DefaultExchange(camelContext);
exchange.getMessage().setBody(body);
transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
assertInstanceOf(String.class, exchange.getMessage().getBody(),
body.getClass().getName());
}
- // Gson cannot write the fields of a stream or a file of the JDK
- for (Object body : List.of(new
ByteArrayInputStream("Test".getBytes(StandardCharsets.UTF_8)), new
File("item.json"))) {
- Exchange exchange = new DefaultExchange(camelContext);
- exchange.getMessage().setBody(body);
- assertThrows(JsonIOException.class, () ->
transformer.transform(exchange.getMessage(), DataType.ANY, DataType.ANY),
- body.getClass().getName());
- }
Exchange exchange = new DefaultExchange(camelContext);
exchange.getMessage().setBody(record);
transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
assertTrue(exchange.getMessage().getBody(String.class).contains("Rajesh"),
exchange.getMessage().getBody(String.class));
// a String is left as it is
- String text = "{\"name\": \"Rajesh\"}";
exchange = new DefaultExchange(camelContext);
- exchange.getMessage().setBody(text);
+ exchange.getMessage().setBody(JSON);
transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
- assertSame(text, exchange.getMessage().getBody());
+ assertSame(JSON, exchange.getMessage().getBody());
+ }
+
+ @Test
+ void aStreamIsReadAsText() throws Exception {
+ // CAMEL-26300: Gson failed with a JsonIOException, as it cannot
reflect into the stream classes of the JDK
+ for (Object body : List.of(new
ByteArrayInputStream(JSON.getBytes(StandardCharsets.UTF_8)), new
StringReader(JSON),
+ new InputStreamCache(JSON.getBytes(StandardCharsets.UTF_8)))) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setBody(body);
+ transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
+ assertEquals(JSON, exchange.getMessage().getBody(),
body.getClass().getName());
+ }
+ }
+
+ @Test
+ void bytesAreReadAsText() throws Exception {
+ // CAMEL-26300: Gson wrote bytes as a Json array of numbers, while the
same Json in a String, a stream or a file
+ // is taken as it is
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setBody(JSON.getBytes(StandardCharsets.UTF_8));
+ transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
+ assertEquals(JSON, exchange.getMessage().getBody());
+
+ // the charset of the exchange applies, as for a stream
+ String json = "{\"name\": \"Andr\u00e9s\"}";
+ exchange = new DefaultExchange(camelContext);
+ exchange.setProperty(Exchange.CHARSET_NAME,
StandardCharsets.ISO_8859_1.name());
+
exchange.getMessage().setBody(json.getBytes(StandardCharsets.ISO_8859_1));
+ transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
+ assertEquals(json, exchange.getMessage().getBody());
+ }
+
+ @Test
+ void aFileIsReadAsText() throws Exception {
+ // CAMEL-26300: Gson failed with a JsonIOException, as it cannot
reflect into java.io.File
+ Path file = Files.writeString(dir.resolve("item.json"), JSON);
+ for (Object body : List.of(file.toFile(), file,
+ FileConsumer.asGenericFile(dir.toString(), file.toFile(),
StandardCharsets.UTF_8.name(), false))) {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setBody(body);
+ transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
+ assertEquals(JSON, exchange.getMessage().getBody(),
body.getClass().getName());
+ }
+ }
+
+ @Test
+ void aFileThatCannotBeReadFails() {
+ File missing = dir.resolve("missing.json").toFile();
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setBody(missing);
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () -> transformer.transform(exchange.getMessage(),
DataType.ANY, DataType.ANY));
+ assertTrue(e.getMessage().startsWith("Failed to read the stream or
file body as text"), e.getMessage());
+ assertFalse(e.getMessage().contains("missing.json"), e.getMessage());
+ assertSame(missing, exchange.getMessage().getBody());
}
}
diff --git
a/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
b/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
index ebcc76e86d44..f13e3f148eb7 100644
---
a/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
+++
b/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
@@ -91,7 +91,8 @@
"CamelAwsSqsIsTruncated": { "index": 10, "kind": "header", "displayName":
"", "group": "listQueues", "label": "listQueues", "required": false,
"javaType": "Boolean", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Whether the response has more results
(is truncated).", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#IS_TRUNCATED" },
"CamelAwsSqsSequenceNumber": { "index": 11, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The sequence number for FIFO queues.",
"constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SEQUENCE_NUMBER" },
"CamelAwsSqsFailedMessageCount": { "index": 12, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of failed messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#FAILED_MESSAGE_COUNT" },
- "CamelAwsSqsSuccessfulMessageCount": { "index": 13, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of successful messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SUCCESSFUL_MESSAGE_COUNT" }
+ "CamelAwsSqsSuccessfulMessageCount": { "index": 13, "kind": "header",
"displayName": "", "group": "sendBatchMessage", "label": "sendBatchMessage",
"required": false, "javaType": "Integer", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
number of successful messages in a batch send operation.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#SUCCESSFUL_MESSAGE_COUNT" },
+ "CamelAwsSqsQueueName": { "index": 14, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The name of the queue the message was received
from.", "constantName":
"org.apache.camel.component.aws2.sqs.Sqs2Constants#QUEUE_NAME" }
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String",
"description": "The body of the SQS message" },
diff --git
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Constants.java
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Constants.java
index f7aad3656a12..db13a85e7bd2 100644
---
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Constants.java
+++
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Constants.java
@@ -70,4 +70,8 @@ public interface Sqs2Constants {
@Metadata(label = "sendBatchMessage",
description = "The number of successful messages in a batch send
operation.", javaType = "Integer")
String SUCCESSFUL_MESSAGE_COUNT = "CamelAwsSqsSuccessfulMessageCount";
+
+ @Metadata(label = "consumer", description = "The name of the queue the
message was received from.",
+ javaType = "String")
+ String QUEUE_NAME = "CamelAwsSqsQueueName";
}
diff --git
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
index a0cb57997e4a..0c3f426dadef 100644
---
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
+++
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
@@ -62,7 +62,9 @@ import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.support.DefaultScheduledPollConsumerScheduler;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.util.CastUtils;
+import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.StringHelper;
import org.apache.camel.util.URISupport;
import org.apache.commons.io.function.IOConsumer;
import org.slf4j.Logger;
@@ -96,10 +98,28 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
private ScheduledExecutorService scheduledExecutor;
private PollingTask pollingTask;
private final String sqsConsumerToString;
+ // the name of the queue the consumer reads, for the CamelAwsSqsQueueName
header
+ private final String readQueueName;
public Sqs2Consumer(Sqs2Endpoint endpoint, Processor processor) {
super(endpoint, processor);
sqsConsumerToString =
"SqsConsumer[%s]".formatted(URISupport.sanitizeUri(endpoint.getEndpointUri()));
+ readQueueName = queueNameOf(endpoint.getConfiguration());
+ }
+
+ /**
+ * The name of the queue the consumer reads: the last part of the path of
the queueUrl option when it is set, as the
+ * other options that name the queue are ignored then, else the queue name
of the endpoint.
+ */
+ private static String queueNameOf(Sqs2Configuration configuration) {
+ String queueUrl = configuration.getQueueUrl();
+ if (ObjectHelper.isNotEmpty(queueUrl)) {
+ String name =
StringHelper.afterLast(FileUtil.stripTrailingSeparator(queueUrl), "/");
+ if (ObjectHelper.isNotEmpty(name)) {
+ return name;
+ }
+ }
+ return configuration.getQueueName();
}
@Override
@@ -311,6 +331,7 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
message.setHeader(Sqs2Constants.MESSAGE_ID, msg.messageId());
message.setHeader(Sqs2Constants.MD5_OF_BODY, msg.md5OfBody());
message.setHeader(Sqs2Constants.RECEIPT_HANDLE, msg.receiptHandle());
+ message.setHeader(Sqs2Constants.QUEUE_NAME, readQueueName);
message.setHeader(Sqs2Constants.ATTRIBUTES, msg.attributes());
message.setHeader(Sqs2Constants.MESSAGE_ATTRIBUTES,
msg.messageAttributes());
diff --git
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformer.java
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformer.java
index 5744fc7b859e..472371cbe3f8 100644
---
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformer.java
+++
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformer.java
@@ -32,7 +32,8 @@ import org.apache.camel.spi.Transformer;
/**
* Output data type represents AWS SQS receive Message response as CloudEvent
V1. The data type sets Camel specific
- * CloudEvent headers on the exchange.
+ * CloudEvent headers on the exchange. The CloudEvent source is the name of
the queue, from the header the aws2-sqs
+ * consumer sets, with the prefix {@code aws.sqs.queue.}.
*/
@DataTypeTransformer(name = "aws2-sqs:application-cloudevents",
description = "Adds CloudEvent headers to the Camel
message with AWS SQS receive message details")
@@ -50,9 +51,9 @@ public class AWS2SQSCloudEventDataTypeTransformer extends
Transformer {
headers.putIfAbsent(CloudEvent.CAMEL_CLOUD_EVENT_VERSION,
cloudEvent.version());
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_TYPE,
"org.apache.camel.event.aws.sqs.receiveMessage");
- if (message.getHeaders().containsKey(Sqs2Constants.RECEIPT_HANDLE)) {
+ if (message.getHeaders().containsKey(Sqs2Constants.QUEUE_NAME)) {
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_SOURCE,
- "aws.sqs.queue." +
message.getHeader(Sqs2Constants.RECEIPT_HANDLE, String.class));
+ "aws.sqs.queue." +
message.getHeader(Sqs2Constants.QUEUE_NAME, String.class));
}
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_SUBJECT,
message.getHeader(Sqs2Constants.MESSAGE_ID, String.class));
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerRouteTest.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerRouteTest.java
new file mode 100644
index 000000000000..9cc15a46fd81
--- /dev/null
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerRouteTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.aws2.sqs.transform;
+
+import java.util.List;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.cloudevents.CloudEvent;
+import org.apache.camel.component.aws2.sqs.AmazonSQSClientMock;
+import org.apache.camel.component.aws2.sqs.Sqs2Constants;
+import org.apache.camel.component.aws2.sqs.Sqs2Consumer;
+import org.apache.camel.component.aws2.sqs.Sqs2Endpoint;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.DataType;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.sqs.model.Message;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * The CloudEvent source of a message of the aws2-sqs consumer, when the
transformer runs in another route, as with the
+ * output data type of a Kamelet: that route gets a copy of the exchange with
the headers of the consumer, and its own
+ * from endpoint.
+ */
+class AWS2SQSCloudEventDataTypeTransformerRouteTest extends CamelTestSupport {
+
+ @BindToRegistry("amazonSQSClient")
+ private final AmazonSQSClientMock client = new AmazonSQSClientMock();
+
+ @Test
+ void theSourceIsTheQueueOfTheConsumer() throws Exception {
+ // CAMEL-26300: the source was built from the receipt handle, which
changes each time the message is received
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello");
+ result.expectedHeaderReceived(Sqs2Constants.QUEUE_NAME, "MyQueue");
+ result.expectedHeaderReceived(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE,
"aws.sqs.queue.MyQueue");
+
+ template.sendBody("direct:start", "Hello");
+
+ MockEndpoint.assertIsSatisfied(context);
+ Exchange received = result.getReceivedExchanges().get(0);
+
assertNotNull(received.getMessage().getHeader(Sqs2Constants.RECEIPT_HANDLE));
+ assertEquals("org.apache.camel.event.aws.sqs.receiveMessage",
+
received.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_TYPE));
+ assertEquals(received.getMessage().getHeader(Sqs2Constants.MESSAGE_ID),
+
received.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SUBJECT));
+ }
+
+ @Test
+ void theSourceIsTheQueueNameOfAnArn() throws Exception {
+ Sqs2Endpoint endpoint = context.getEndpoint(
+
"aws2-sqs://arn:aws:sqs:eu-west-1:123456789012:MyArnQueue?amazonSQSClient=#amazonSQSClient",
+ Sqs2Endpoint.class);
+ Sqs2Consumer consumer = (Sqs2Consumer)
endpoint.createConsumer(exchange -> {
+ });
+ Exchange exchange = consumer.createExchange(
+
Message.builder().body("Hello").messageId("1234").receiptHandle("AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a").build());
+ assertEquals("MyArnQueue",
exchange.getMessage().getHeader(Sqs2Constants.QUEUE_NAME));
+
+ new
AWS2SQSCloudEventDataTypeTransformer().transform(exchange.getMessage(),
DataType.ANY, DataType.ANY);
+ assertEquals("aws.sqs.queue.MyArnQueue",
exchange.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE));
+ }
+
+ @Test
+ void theSourceIsTheQueueOfTheQueueUrl() throws Exception {
+ // the queueUrl option names the queue that is read, and the other
options that name a queue are ignored then
+ for (String queueUrl :
List.of("https://sqs.eu-west-1.amazonaws.com/123456789012/OtherQueue";,
+ "http://localhost:4566/000000000000/OtherQueue/";)) {
+ Sqs2Endpoint endpoint = context.getEndpoint(
+
"aws2-sqs://MyQueue?amazonSQSClient=#amazonSQSClient&queueUrl=RAW(" + queueUrl
+ ")", Sqs2Endpoint.class);
+ Sqs2Consumer consumer = (Sqs2Consumer)
endpoint.createConsumer(exchange -> {
+ });
+ Exchange exchange = consumer.createExchange(
+
Message.builder().body("Hello").messageId("1234").receiptHandle("AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a")
+ .build());
+ assertEquals("OtherQueue",
exchange.getMessage().getHeader(Sqs2Constants.QUEUE_NAME), queueUrl);
+
+ new
AWS2SQSCloudEventDataTypeTransformer().transform(exchange.getMessage(),
DataType.ANY, DataType.ANY);
+ assertEquals("aws.sqs.queue.OtherQueue",
exchange.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE),
+ queueUrl);
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").to("aws2-sqs://MyQueue?amazonSQSClient=#amazonSQSClient");
+
+ // seda hands a copy of the exchange to the next route, as the
sink of a Kamelet does
+
from("aws2-sqs://MyQueue?amazonSQSClient=#amazonSQSClient").to("seda:next");
+
+ from("seda:next")
+ .transformDataType("aws2-sqs:application-cloudevents")
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerTest.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerTest.java
index 882cc1b8c817..133a8e2c415b 100644
---
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerTest.java
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/transform/AWS2SQSCloudEventDataTypeTransformerTest.java
@@ -36,6 +36,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
public class AWS2SQSCloudEventDataTypeTransformerTest {
@@ -48,7 +49,8 @@ public class AWS2SQSCloudEventDataTypeTransformerTest {
void shouldMapToCloudEvent() throws Exception {
Exchange exchange = new DefaultExchange(camelContext);
- exchange.getMessage().setHeader(Sqs2Constants.RECEIPT_HANDLE,
"myQueue");
+ exchange.getMessage().setHeader(Sqs2Constants.QUEUE_NAME, "myQueue");
+ exchange.getMessage().setHeader(Sqs2Constants.RECEIPT_HANDLE,
"AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a");
exchange.getMessage().setHeader(Sqs2Constants.MESSAGE_ID, "1234");
exchange.getMessage().setBody(new
ByteArrayInputStream("Test1".getBytes(StandardCharsets.UTF_8)));
transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
@@ -61,6 +63,18 @@ public class AWS2SQSCloudEventDataTypeTransformerTest {
assertEquals("aws.sqs.queue.myQueue",
exchange.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE));
}
+ @Test
+ void theSourceIsNotTheReceiptHandle() throws Exception {
+ // CAMEL-26300: the receipt handle changes each time a message is
received, so it names no source
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setHeader(Sqs2Constants.RECEIPT_HANDLE,
"AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a");
+ exchange.getMessage().setHeader(Sqs2Constants.MESSAGE_ID, "1234");
+ transformer.transform(exchange.getMessage(), DataType.ANY,
DataType.ANY);
+
+
assertNull(exchange.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE));
+ assertEquals("1234",
exchange.getMessage().getHeader(CloudEvent.CAMEL_CLOUD_EVENT_SUBJECT));
+ }
+
@Test
public void shouldLookupDataTypeTransformer() throws Exception {
Transformer transformer = camelContext.getTransformerRegistry()
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index a0c7ff895e8c..c5120c17c7b1 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -5339,6 +5339,26 @@ The `SAVE_DOCUMENT` operation now inserts a `Collection`
body, such as a `List`
it does with the `CamelArangoDbMultiInsert` header, and the body after it is a
`MultiDocumentEntity`. Before, without
the header, ArangoDB inserted the documents and the exchange then failed, so a
redelivery inserted them again.
+=== camel-aws2-sqs - the CloudEvent source is the queue name
+
+The `aws2-sqs:application-cloudevents` data type transformer now sets the
`CamelCloudEventSource` header to
+`aws.sqs.queue.` followed by the name of the queue the message was received
from. Before, it used the receipt handle
+of the message, which changes each time a message is received, so every event
had a source of its own.
+
+The queue name comes from the new `CamelAwsSqsQueueName` header, which the
`aws2-sqs` consumer now sets on each
+message. When the endpoint names the queue by its ARN, the header holds the
queue name part of the ARN, and when the
+`queueUrl` option is set, the last part of its path, as that is the queue the
consumer reads. A message without this
+header, such as one that does not come from the `aws2-sqs` consumer, gets no
CloudEvent source from the transformer,
+also when it has a `CamelAwsSqsReceiptHandle` header. A route that relies on
the old source can set the
+`CamelCloudEventSource` header itself after the transformer.
+
+=== camel-aws2-ddb - the x-struct transformer reads bytes as JSON text
+
+The `aws2-ddb:application-x-struct` data type transformer now reads a `byte[]`
body as JSON text, with the charset of
+the exchange, as it reads a stream or a file and takes a String as it is, so
the same JSON gives the same String
+whatever the type of the body. Before, Gson wrote the bytes as a JSON array of
numbers. A stream or a file body, which
+failed before, is now read as text as well.
+
=== camel-ftp, camel-sftp, camel-ftps, camel-mina-sftp - sendNoop also applies
to the consumer
The `sendNoop` option now also applies to the check a consumer makes before
each poll, not only to the pre-write check
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Sqs2EndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Sqs2EndpointBuilderFactory.java
index 3a03b02ab6a0..6ec020a48143 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Sqs2EndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Sqs2EndpointBuilderFactory.java
@@ -3507,6 +3507,18 @@ public interface Sqs2EndpointBuilderFactory {
public String awsSqsSuccessfulMessageCount() {
return "CamelAwsSqsSuccessfulMessageCount";
}
+ /**
+ * The name of the queue the message was received from.
+ *
+ * The option is a: {@code String} type.
+ *
+ * Group: consumer
+ *
+ * @return the name of the header {@code AwsSqsQueueName}.
+ */
+ public String awsSqsQueueName() {
+ return "CamelAwsSqsQueueName";
+ }
}
static Sqs2EndpointBuilder endpointBuilder(String componentName, String
path) {
class Sqs2EndpointBuilderImpl extends AbstractEndpointBuilder
implements Sqs2EndpointBuilder, AdvancedSqs2EndpointBuilder {
