This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26263-dapr-secret-headers in repository https://gitbox.apache.org/repos/asf/camel.git
commit 1daea74ef9fdba7b2004d31a4139e749dea179cd Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 15:36:33 2026 +0100 CAMEL-26263: camel-dapr - register the secret operation, keep the message headers, and send text and stream bodies as bytes The secret operation had no handler, so every dapr:secret call failed with No handler for operation secret. The DaprSecretHandler is now registered. The producer replaced all the message headers with the headers of the Dapr response. It now keeps the headers and adds the response headers. The invokeService, pubSub, invokeBinding and state save operations gave the body to the Dapr client as is, whose JSON serializer sent a String as a quoted JSON string and failed for a stream. A text or stream body is now converted to bytes, which the client sends as they are. 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 | 4 + .../org/apache/camel/component/dapr/dapr.json | 2 +- .../camel-dapr/src/main/docs/dapr-component.adoc | 4 + .../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, 273 insertions(+), 12 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..03286535ec8a 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. 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..03286535ec8a 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. 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 7c82ebd18930..3f8f12ed2482 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 @@ -828,6 +828,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-aws - the HTTP client is no longer looked up on the classpath The camel-aws components now set the HTTP client of the AWS SDK clients they create for their endpoints: the Apache 5
