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 a14dd5068f17 CAMEL-26263: camel-dapr - register the secret operation,
keep the message headers, and send text and stream bodies as bytes (#29105)
a14dd5068f17 is described below
commit a14dd5068f179dd51dbc05ababf49eee772f14d1
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Nov 4 20:49:47 2026 +0100
CAMEL-26263: camel-dapr - register the secret operation, keep the message
headers, and send text and stream bodies as bytes (#29105)
- CAMEL-26263: camel-dapr - register the secret operation, keep the message
headers, and send text and stream bodies as bytes
- CAMEL-26263: camel-dapr - docs - the secret examples no longer log the
secret body
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/dapr.json | 2 +-
.../apache/camel/catalog/docs/dapr-component.adoc | 16 +-
.../org/apache/camel/component/dapr/dapr.json | 2 +-
.../camel-dapr/src/main/docs/dapr-component.adoc | 16 +-
.../apache/camel/component/dapr/DaprEndpoint.java | 11 +-
.../apache/camel/component/dapr/DaprProducer.java | 3 +-
.../dapr/operations/DaprInvokeBindingHandler.java | 2 +-
.../dapr/operations/DaprOperationHandler.java | 20 ++
.../dapr/operations/DaprOperationManager.java | 1 +
.../dapr/operations/DaprPubSubHandler.java | 2 +-
.../operations/DaprServiceInvocationHandler.java | 2 +-
.../dapr/operations/DaprStateHandler.java | 2 +-
.../camel/component/dapr/DaprProducerTest.java | 209 +++++++++++++++++++++
.../component/dapr/operations/DaprStateTest.java | 7 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 ++
15 files changed, 285 insertions(+), 24 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/dapr.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/dapr.json
index a5886881c019..5828a5b38408 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/dapr.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/dapr.json
@@ -127,7 +127,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "map" ], "byOption": [ { "option":
"operation", "value": "pubSub", "carriers": [ "bytes" ], "javaType": "byte[]"
}, { "option": "operation", "value": "configuration", "carriers": [ "map" ],
"javaType": "java.util.Map<java.lang.String, java.lang.String>" } ],
"description": "For pubSub the data of the cloud event as bytes, with its id,
type, source and content type in headers. For configuration a Map of the
changed keys and their values, with th [...]
- "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"pojo", "null" ], "description": "The data of invokeService, pubSub,
invokeBinding, state save and the workflow input, event and output, given to
the Dapr client as is: bytes are sent as they are, other bodies are serialized
as JSON (text as a JSON string), and a stream fails: convert it to bytes first.
State save needs a body; the other operations do not read it" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "value", "map",
"list", "pojo", "null" ], "description": "The data of invokeService, pubSub,
invokeBinding, state save and the workflow input, event and output. For
invokeService, pubSub, invokeBinding and state save, text and streams are
converted to bytes, bytes are sent as they are, and other bodies are serialized
as JSON. The workflow operations give the body to the workflow client as is,
which serializes it as JSON, so a s [...]
"producerOut": { "carriers": [ "bytes", "pojo", "list", "map", "value",
"unchanged", "null" ], "byOption": [ { "option": "operation", "value":
"invokeService", "carriers": [ "bytes", "null" ] }, { "option": "operation",
"value": "invokeBinding", "carriers": [ "bytes", "null" ] }, { "option":
"operation", "value": "pubSub", "carriers": [ "pojo" ], "javaType":
"io.dapr.client.domain.PublishEventRequest" }, { "option": "operation",
"value": "state", "carriers": [ "pojo", "list" ] }, { " [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
index efe5e2a7d761..cce7ea2f74c4 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
@@ -33,6 +33,10 @@ include::partial$component-endpoint-headers.adoc[]
Camel Dapr component provides a wide range of operations on the producer side:
+The producer keeps the headers of the message, and adds the headers of the
Dapr response, such as `CamelDaprRawConfigResponse` for the configuration
operation.
+
+The `invokeService`, `pubSub`, `invokeBinding` and `state` `save` operations
send the message body as the data: a text or stream body is converted to bytes
and sent as it is, bytes are sent as they are, and other bodies, such as a
`Map` or a POJO, are serialized as JSON. For `pubSub`, set the content type of
the data with the `contentType` option or the `CamelDaprContentType` header.
+
=== 1. *invokeService*:
Using service invocation, your application can reliably and securely
communicate with other applications using the standard HTTP protocol. For this
operation `serviceToInvoke` and `methodToInvoke` are mandatory.
@@ -431,7 +435,7 @@ Java::
----
from("direct:start")
.to("dapr:secret?secretStore=myStore&key=mySecret")
- .log("Received secret from Dapr: ${body}")
+ .log("Received secret mySecret from Dapr")
.to("mock:result");
----
@@ -442,7 +446,7 @@ XML::
<route>
<from uri="direct:start"/>
<to uri="dapr:secret?secretStore=myStore&key=mySecret"/>
- <log message="Received secret from Dapr: ${body}"/>
+ <log message="Received secret mySecret from Dapr"/>
<to uri="mock:result"/>
</route>
----
@@ -460,7 +464,7 @@ YAML::
parameters:
secretStore: myStore
key: mySecret
- - log: "Received secret from Dapr: ${body}"
+ - log: "Received secret mySecret from Dapr"
- to:
uri: mock:result
----
@@ -477,7 +481,7 @@ Java::
from("direct:start")
// when no secret key is passed, bulk retrieval is performed on secret store
.to("dapr:secret?secretStore=myStore")
- .log("Received bulk secrets from Dapr: ${body}")
+ .log("Received ${body.size()} secrets from Dapr")
.to("mock:result");
----
@@ -488,7 +492,7 @@ XML::
<route>
<from uri="direct:start"/>
<to uri="dapr:secret?secretStore=myStore"/>
- <log message="Received bulk secrets from Dapr: ${body}"/>
+ <log message="Received ${body.size()} secrets from Dapr"/>
<to uri="mock:result"/>
</route>
----
@@ -505,7 +509,7 @@ YAML::
uri: dapr:secret
parameters:
secretStore: myStore
- - log: "Received bulk secrets from Dapr: ${body}"
+ - log: "Received ${body.size()} secrets from Dapr"
- to:
uri: mock:result
----
diff --git
a/components/camel-dapr/src/generated/resources/META-INF/org/apache/camel/component/dapr/dapr.json
b/components/camel-dapr/src/generated/resources/META-INF/org/apache/camel/component/dapr/dapr.json
index a5886881c019..5828a5b38408 100644
---
a/components/camel-dapr/src/generated/resources/META-INF/org/apache/camel/component/dapr/dapr.json
+++
b/components/camel-dapr/src/generated/resources/META-INF/org/apache/camel/component/dapr/dapr.json
@@ -127,7 +127,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes", "map" ], "byOption": [ { "option":
"operation", "value": "pubSub", "carriers": [ "bytes" ], "javaType": "byte[]"
}, { "option": "operation", "value": "configuration", "carriers": [ "map" ],
"javaType": "java.util.Map<java.lang.String, java.lang.String>" } ],
"description": "For pubSub the data of the cloud event as bytes, with its id,
type, source and content type in headers. For configuration a Map of the
changed keys and their values, with th [...]
- "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"pojo", "null" ], "description": "The data of invokeService, pubSub,
invokeBinding, state save and the workflow input, event and output, given to
the Dapr client as is: bytes are sent as they are, other bodies are serialized
as JSON (text as a JSON string), and a stream fails: convert it to bytes first.
State save needs a body; the other operations do not read it" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "value", "map",
"list", "pojo", "null" ], "description": "The data of invokeService, pubSub,
invokeBinding, state save and the workflow input, event and output. For
invokeService, pubSub, invokeBinding and state save, text and streams are
converted to bytes, bytes are sent as they are, and other bodies are serialized
as JSON. The workflow operations give the body to the workflow client as is,
which serializes it as JSON, so a s [...]
"producerOut": { "carriers": [ "bytes", "pojo", "list", "map", "value",
"unchanged", "null" ], "byOption": [ { "option": "operation", "value":
"invokeService", "carriers": [ "bytes", "null" ] }, { "option": "operation",
"value": "invokeBinding", "carriers": [ "bytes", "null" ] }, { "option":
"operation", "value": "pubSub", "carriers": [ "pojo" ], "javaType":
"io.dapr.client.domain.PublishEventRequest" }, { "option": "operation",
"value": "state", "carriers": [ "pojo", "list" ] }, { " [...]
},
"properties": {
diff --git a/components/camel-dapr/src/main/docs/dapr-component.adoc
b/components/camel-dapr/src/main/docs/dapr-component.adoc
index efe5e2a7d761..cce7ea2f74c4 100644
--- a/components/camel-dapr/src/main/docs/dapr-component.adoc
+++ b/components/camel-dapr/src/main/docs/dapr-component.adoc
@@ -33,6 +33,10 @@ include::partial$component-endpoint-headers.adoc[]
Camel Dapr component provides a wide range of operations on the producer side:
+The producer keeps the headers of the message, and adds the headers of the
Dapr response, such as `CamelDaprRawConfigResponse` for the configuration
operation.
+
+The `invokeService`, `pubSub`, `invokeBinding` and `state` `save` operations
send the message body as the data: a text or stream body is converted to bytes
and sent as it is, bytes are sent as they are, and other bodies, such as a
`Map` or a POJO, are serialized as JSON. For `pubSub`, set the content type of
the data with the `contentType` option or the `CamelDaprContentType` header.
+
=== 1. *invokeService*:
Using service invocation, your application can reliably and securely
communicate with other applications using the standard HTTP protocol. For this
operation `serviceToInvoke` and `methodToInvoke` are mandatory.
@@ -431,7 +435,7 @@ Java::
----
from("direct:start")
.to("dapr:secret?secretStore=myStore&key=mySecret")
- .log("Received secret from Dapr: ${body}")
+ .log("Received secret mySecret from Dapr")
.to("mock:result");
----
@@ -442,7 +446,7 @@ XML::
<route>
<from uri="direct:start"/>
<to uri="dapr:secret?secretStore=myStore&key=mySecret"/>
- <log message="Received secret from Dapr: ${body}"/>
+ <log message="Received secret mySecret from Dapr"/>
<to uri="mock:result"/>
</route>
----
@@ -460,7 +464,7 @@ YAML::
parameters:
secretStore: myStore
key: mySecret
- - log: "Received secret from Dapr: ${body}"
+ - log: "Received secret mySecret from Dapr"
- to:
uri: mock:result
----
@@ -477,7 +481,7 @@ Java::
from("direct:start")
// when no secret key is passed, bulk retrieval is performed on secret store
.to("dapr:secret?secretStore=myStore")
- .log("Received bulk secrets from Dapr: ${body}")
+ .log("Received ${body.size()} secrets from Dapr")
.to("mock:result");
----
@@ -488,7 +492,7 @@ XML::
<route>
<from uri="direct:start"/>
<to uri="dapr:secret?secretStore=myStore"/>
- <log message="Received bulk secrets from Dapr: ${body}"/>
+ <log message="Received ${body.size()} secrets from Dapr"/>
<to uri="mock:result"/>
</route>
----
@@ -505,7 +509,7 @@ YAML::
uri: dapr:secret
parameters:
secretStore: myStore
- - log: "Received bulk secrets from Dapr: ${body}"
+ - log: "Received ${body.size()} secrets from Dapr"
- to:
uri: mock:result
----
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprEndpoint.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprEndpoint.java
index ce37787c4661..b4cb04f22f49 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprEndpoint.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprEndpoint.java
@@ -54,11 +54,14 @@ import org.slf4j.LoggerFactory;
+ " their values, with the
ConfigurationItems in the CamelDaprRawConfigResponse"
+ " header"),
producerIn = @BodyType(value = {
- Carrier.TEXT, Carrier.BYTES, Carrier.VALUE, Carrier.MAP,
Carrier.LIST, Carrier.POJO, Carrier.NULL },
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.VALUE,
Carrier.MAP, Carrier.LIST, Carrier.POJO,
+ Carrier.NULL },
description = "The data of invokeService,
pubSub, invokeBinding, state save and the workflow"
- + " input, event and output,
given to the Dapr client as is: bytes are sent as"
- + " they are, other bodies are
serialized as JSON (text as a JSON string), and a"
- + " stream fails: convert it
to bytes first. State save needs a body; the other"
+ + " input, event and output.
For invokeService, pubSub, invokeBinding and state"
+ + " save, text and streams are
converted to bytes, bytes are sent as they are,"
+ + " and other bodies are
serialized as JSON. The workflow operations give the"
+ + " body to the workflow
client as is, which serializes it as JSON, so a stream"
+ + " fails there: convert it
first. State save needs a body; the other"
+ " operations do not read
it"),
producerOut = @BodyType(value = {
Carrier.BYTES, Carrier.POJO, Carrier.LIST, Carrier.MAP,
Carrier.VALUE, Carrier.UNCHANGED, Carrier.NULL },
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprProducer.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprProducer.java
index c96baac8ef70..db822d3b256d 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprProducer.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/DaprProducer.java
@@ -45,7 +45,8 @@ public class DaprProducer extends DefaultProducer {
private void setResponse(Exchange exchange, DaprOperationResponse
response) {
if (response != null) {
exchange.getMessage().setBody(response.getBody());
- exchange.getMessage().setHeaders(response.getHeaders());
+ // keep the headers of the message, and add the headers of the
response
+ exchange.getMessage().getHeaders().putAll(response.getHeaders());
}
}
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprInvokeBindingHandler.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprInvokeBindingHandler.java
index 1f788e44df02..be48e0c7ef44 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprInvokeBindingHandler.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprInvokeBindingHandler.java
@@ -38,7 +38,7 @@ public class DaprInvokeBindingHandler implements
DaprOperationHandler {
@Override
public DaprOperationResponse handle(Exchange exchange) {
- Object payload = exchange.getIn().getBody();
+ Object payload = DaprOperationHandler.getPayload(exchange);
String bindingName =
configurationOptionsProxy.getBindingName(exchange);
String bindingOperation =
configurationOptionsProxy.getBindingOperation(exchange);
Map<String, String> metadata =
configurationOptionsProxy.getMetadata(exchange);
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationHandler.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationHandler.java
index 68e3bfeb25c4..332d719658d6 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationHandler.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationHandler.java
@@ -16,10 +16,30 @@
*/
package org.apache.camel.component.dapr.operations;
+import java.io.InputStream;
+import java.io.Reader;
+
import org.apache.camel.Exchange;
+import org.apache.camel.StreamCache;
public interface DaprOperationHandler {
+ /**
+ * The message body to give to the Dapr client as data. The client sends
bytes as they are, and serializes any other
+ * body as JSON, so text would be sent as a JSON string (with quotes) and
a stream would fail. A text or stream body
+ * is therefore converted to bytes first; other bodies are returned as
they are.
+ *
+ * @param exchange the current exchange
+ * @return the data to give to the Dapr client
+ */
+ static Object getPayload(Exchange exchange) {
+ Object body = exchange.getIn().getBody();
+ if (body instanceof String || body instanceof InputStream || body
instanceof Reader || body instanceof StreamCache) {
+ return exchange.getIn().getBody(byte[].class);
+ }
+ return body;
+ }
+
/**
* Execute the configured operation on this exchange.
*
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationManager.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationManager.java
index a089710e60f7..318e01e1293c 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationManager.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprOperationManager.java
@@ -35,6 +35,7 @@ public class DaprOperationManager {
handlerMap.put(DaprOperation.state, new
DaprStateHandler(configurationOptionsProxy, endpoint));
handlerMap.put(DaprOperation.pubSub, new
DaprPubSubHandler(configurationOptionsProxy, endpoint));
handlerMap.put(DaprOperation.invokeBinding, new
DaprInvokeBindingHandler(configurationOptionsProxy, endpoint));
+ handlerMap.put(DaprOperation.secret, new
DaprSecretHandler(configurationOptionsProxy, endpoint));
handlerMap.put(DaprOperation.configuration, new
DaprConfigurationHandler(configurationOptionsProxy, endpoint));
handlerMap.put(DaprOperation.lock, new
DaprLockHandler(configurationOptionsProxy, endpoint));
handlerMap.put(DaprOperation.workflow, new
DaprWorkflowHandler(configurationOptionsProxy, endpoint));
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprPubSubHandler.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprPubSubHandler.java
index 2f7dd8af2c49..4a68e2c90a42 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprPubSubHandler.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprPubSubHandler.java
@@ -37,7 +37,7 @@ public class DaprPubSubHandler implements
DaprOperationHandler {
@Override
public DaprOperationResponse handle(Exchange exchange) {
- Object payload = exchange.getIn().getBody();
+ Object payload = DaprOperationHandler.getPayload(exchange);
String pubSubName = configurationOptionsProxy.getPubSubName(exchange);
String topic = configurationOptionsProxy.getTopic(exchange);
String contentType =
configurationOptionsProxy.getContentType(exchange);
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprServiceInvocationHandler.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprServiceInvocationHandler.java
index e41a5b5c8fb0..7d6a38e6008d 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprServiceInvocationHandler.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprServiceInvocationHandler.java
@@ -39,7 +39,7 @@ public class DaprServiceInvocationHandler implements
DaprOperationHandler {
@Override
public DaprOperationResponse handle(Exchange exchange) {
- Object payload = exchange.getIn().getBody();
+ Object payload = DaprOperationHandler.getPayload(exchange);
HttpExtension httpExtension = initHttpExtension(exchange);
String service =
configurationOptionsProxy.getServiceToInvoke(exchange);
String method = configurationOptionsProxy.getMethodToInvoke(exchange);
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprStateHandler.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprStateHandler.java
index d9ea0370c855..2b5e69029f32 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprStateHandler.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/operations/DaprStateHandler.java
@@ -69,7 +69,7 @@ public class DaprStateHandler implements DaprOperationHandler
{
}
private DaprOperationResponse saveState(Exchange exchange, DaprClient
client) {
- Object payload = exchange.getIn().getBody();
+ Object payload = DaprOperationHandler.getPayload(exchange);
String stateStore = configurationOptionsProxy.getStateStore(exchange);
String key = configurationOptionsProxy.getKey(exchange);
String eTag = configurationOptionsProxy.getETag(exchange);
diff --git
a/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/DaprProducerTest.java
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/DaprProducerTest.java
new file mode 100644
index 000000000000..6397e71a2f0f
--- /dev/null
+++
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/DaprProducerTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.dapr;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import io.dapr.client.DaprClient;
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.ConfigurationItem;
+import io.dapr.client.domain.GetBulkSecretRequest;
+import io.dapr.client.domain.GetConfigurationRequest;
+import io.dapr.client.domain.GetSecretRequest;
+import io.dapr.client.domain.HttpExtension;
+import io.dapr.client.domain.InvokeBindingRequest;
+import io.dapr.client.domain.PublishEventRequest;
+import io.dapr.client.domain.StateOptions;
+import io.dapr.utils.TypeRef;
+import io.dapr.workflows.client.DaprWorkflowClient;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Mono;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The producer, called through the endpoint with a mocked Dapr client.
+ */
+class DaprProducerTest extends CamelTestSupport {
+
+ private static final String CLIENTS =
"client=#myClient&previewClient=#myPreviewClient&workflowClient=#myWorkflowClient";
+
+ private final DaprClient client = mock(DaprClient.class);
+ private final DaprPreviewClient previewClient =
mock(DaprPreviewClient.class);
+ private final DaprWorkflowClient workflowClient =
mock(DaprWorkflowClient.class);
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("myClient", client);
+ registry.bind("myPreviewClient", previewClient);
+ registry.bind("myWorkflowClient", workflowClient);
+ }
+
+ @Test
+ void getsSecret() {
+ Map<String, String> secret = Map.of("mySecret", "myValue");
+
when(client.getSecret(any(GetSecretRequest.class))).thenReturn(Mono.just(secret));
+
+ Exchange exchange =
template.request("dapr:secret?secretStore=myStore&key=mySecret&" + CLIENTS,
+ e -> e.getIn().setHeader("foo", "bar"));
+
+ assertNull(exchange.getException());
+ assertEquals(secret, exchange.getMessage().getBody());
+ assertEquals("bar", exchange.getMessage().getHeader("foo"));
+ ArgumentCaptor<GetSecretRequest> request =
ArgumentCaptor.forClass(GetSecretRequest.class);
+ verify(client).getSecret(request.capture());
+ assertEquals("myStore", request.getValue().getStoreName());
+ assertEquals("mySecret", request.getValue().getKey());
+ }
+
+ @Test
+ void getsBulkSecret() {
+ Map<String, Map<String, String>> secrets = Map.of("mySecret",
Map.of("mySecret", "myValue"));
+
when(client.getBulkSecret(any(GetBulkSecretRequest.class))).thenReturn(Mono.just(secrets));
+
+ Exchange exchange =
template.request("dapr:secret?secretStore=myStore&" + CLIENTS, e -> {
+ });
+
+ assertNull(exchange.getException());
+ assertEquals(secrets, exchange.getMessage().getBody());
+ }
+
+ @Test
+ void keepsTheHeadersOfTheMessageAndAddsTheResponseHeaders() {
+ Map<String, ConfigurationItem> config = Map.of("myKey", new
ConfigurationItem("myKey", "myValue", "1"));
+
when(client.getConfiguration(any(GetConfigurationRequest.class))).thenReturn(Mono.just(config));
+
+ Exchange exchange =
template.request("dapr:configuration?configStore=myStore&configKeys=myKey&" +
CLIENTS,
+ e -> e.getIn().setHeader("foo", "bar"));
+
+ assertNull(exchange.getException());
+ assertEquals(Map.of("myKey", "myValue"),
exchange.getMessage().getBody());
+ assertEquals("bar", exchange.getMessage().getHeader("foo"));
+ assertSame(config,
exchange.getMessage().getHeader(DaprConstants.RAW_CONFIG_RESPONSE));
+
+ // the headers can still be changed afterwards
+ exchange.getMessage().setHeader("other", "value");
+ assertEquals("value", exchange.getMessage().getHeader("other"));
+ }
+
+ @Test
+ void sendsTextBodyAsBytesToInvokeService() {
+ when(client.invokeMethod(eq("myService"), eq("myMethod"), any(),
any(HttpExtension.class), eq(byte[].class)))
+ .thenReturn(Mono.just("ok".getBytes(StandardCharsets.UTF_8)));
+
+ Exchange exchange = template.request(
+
"dapr:invokeService?serviceToInvoke=myService&methodToInvoke=myMethod&" +
CLIENTS,
+ e -> e.getIn().setBody("hello"));
+
+ assertNull(exchange.getException());
+ ArgumentCaptor<Object> payload = ArgumentCaptor.forClass(Object.class);
+ verify(client).invokeMethod(eq("myService"), eq("myMethod"),
payload.capture(), any(HttpExtension.class),
+ eq(byte[].class));
+ assertBytes("hello", payload.getValue());
+ }
+
+ @Test
+ void sendsStreamBodyAsBytesToInvokeService() {
+ when(client.invokeMethod(eq("myService"), eq("myMethod"), any(),
any(HttpExtension.class), eq(byte[].class)))
+ .thenReturn(Mono.just("ok".getBytes(StandardCharsets.UTF_8)));
+
+ Exchange exchange = template.request(
+
"dapr:invokeService?serviceToInvoke=myService&methodToInvoke=myMethod&" +
CLIENTS,
+ e -> e.getIn().setBody(new
ByteArrayInputStream("hello".getBytes(StandardCharsets.UTF_8))));
+
+ assertNull(exchange.getException());
+ ArgumentCaptor<Object> payload = ArgumentCaptor.forClass(Object.class);
+ verify(client).invokeMethod(eq("myService"), eq("myMethod"),
payload.capture(), any(HttpExtension.class),
+ eq(byte[].class));
+ assertBytes("hello", payload.getValue());
+ }
+
+ @Test
+ void sendsMapBodyAsItIsToInvokeService() {
+ when(client.invokeMethod(eq("myService"), eq("myMethod"), any(),
any(HttpExtension.class), eq(byte[].class)))
+ .thenReturn(Mono.just("ok".getBytes(StandardCharsets.UTF_8)));
+ Map<String, String> body = Map.of("name", "value");
+
+ Exchange exchange = template.request(
+
"dapr:invokeService?serviceToInvoke=myService&methodToInvoke=myMethod&" +
CLIENTS,
+ e -> e.getIn().setBody(body));
+
+ assertNull(exchange.getException());
+ verify(client).invokeMethod(eq("myService"), eq("myMethod"), eq(body),
any(HttpExtension.class),
+ eq(byte[].class));
+ }
+
+ @Test
+ void sendsTextBodyAsBytesToPubSub() {
+
when(client.publishEvent(any(PublishEventRequest.class))).thenReturn(Mono.empty());
+
+ Exchange exchange =
template.request("dapr:pubSub?pubSubName=myPubSub&topic=myTopic&" + CLIENTS,
+ e -> e.getIn().setBody("hello"));
+
+ assertNull(exchange.getException());
+ ArgumentCaptor<PublishEventRequest> request =
ArgumentCaptor.forClass(PublishEventRequest.class);
+ verify(client).publishEvent(request.capture());
+ assertBytes("hello", request.getValue().getData());
+ }
+
+ @Test
+ void sendsTextBodyAsBytesToInvokeBinding() {
+ when(client.invokeBinding(any(InvokeBindingRequest.class),
any(TypeRef.class))).thenReturn(Mono.empty());
+
+ Exchange exchange =
template.request("dapr:invokeBinding?bindingName=myBinding&bindingOperation=create&"
+ CLIENTS,
+ e -> e.getIn().setBody("hello"));
+
+ assertNull(exchange.getException());
+ ArgumentCaptor<InvokeBindingRequest> request =
ArgumentCaptor.forClass(InvokeBindingRequest.class);
+ verify(client).invokeBinding(request.capture(), any(TypeRef.class));
+ assertBytes("hello", request.getValue().getData());
+ }
+
+ @Test
+ void savesTextBodyAsBytes() {
+ when(client.saveState(eq("myStore"), eq("myKey"), isNull(), any(),
(StateOptions) isNull())).thenReturn(Mono.empty());
+
+ Exchange exchange =
template.request("dapr:state?stateOperation=save&stateStore=myStore&key=myKey&"
+ CLIENTS,
+ e -> e.getIn().setBody("hello"));
+
+ assertNull(exchange.getException());
+ ArgumentCaptor<Object> payload = ArgumentCaptor.forClass(Object.class);
+ verify(client).saveState(eq("myStore"), eq("myKey"), isNull(),
payload.capture(), (StateOptions) isNull());
+ assertBytes("hello", payload.getValue());
+ }
+
+ private static void assertBytes(String expected, Object actual) {
+ byte[] bytes = assertInstanceOf(byte[].class, actual);
+ assertArrayEquals(expected.getBytes(StandardCharsets.UTF_8), bytes);
+ }
+}
diff --git
a/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/operations/DaprStateTest.java
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/operations/DaprStateTest.java
index f2354366d799..ea6b655b6d2b 100644
---
a/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/operations/DaprStateTest.java
+++
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/operations/DaprStateTest.java
@@ -44,8 +44,10 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import reactor.core.publisher.Mono;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -82,7 +84,10 @@ public class DaprStateTest extends CamelTestSupport {
final DaprOperationResponse operationResponse =
operation.handle(exchange);
assertNotNull(operationResponse);
- assertEquals(new State<>("myKey", "myBody", null, null),
operationResponse.getBody());
+ // the text body is saved as its bytes
+ State<?> state = assertInstanceOf(State.class,
operationResponse.getBody());
+ assertEquals("myKey", state.getKey());
+ assertArrayEquals("myBody".getBytes(StandardCharsets.UTF_8), (byte[])
state.getValue());
}
@Test
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 6bd95d30e24a..aa2c10c7a18d 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
@@ -925,6 +925,20 @@ receives such a body as a `String`.
A `null` body is now published as a JSON `null` that the subscribed clients
receive. Before, the message had no data,
so the clients took it for the reply of a publish and never received it, while
the exchange succeeded.
+=== camel-dapr - the producer keeps the message headers, and sends text and
stream bodies as bytes
+
+The `dapr` producer replaced all the headers of the message with the headers
of the Dapr response, so every header set
+before the call was gone afterwards. It now keeps the headers of the message
and adds the headers of the response. A
+`CamelDapr*` header set before a `dapr` call therefore also applies to the
next `dapr` call of the route; remove it
+when the next call must use the options of its endpoint.
+
+The `invokeService`, `pubSub`, `invokeBinding` and `state` `save` operations
now convert a text body (`String`) and a
+stream body (`InputStream`, `Reader`) to bytes, which the Dapr client sends as
they are. Before, the Dapr client
+serialized a text body as a JSON string, so `hello` was sent as `"hello"` with
the quotes, and a JSON text was sent
+as a quoted, escaped JSON string; a stream body failed. A route that relied on
the quoted text must now send the JSON
+it wants, for example with `marshal().json()`. Other bodies, such as a `Map`
or a POJO, are still serialized as JSON.
+The workflow operations give the body to the workflow client as before.
+
=== camel-consul - no Optional in the body, and the CamelConsulResult header
of every action
The `GET_SESSIONS` action of `consul:kv`, the `RENEW` action of
`consul:session` and the `GET` action of
