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 03deff1be5d0 CAMEL-25543: camel-dapr - Send the saveBulk and
executeTransaction state requests (#27691)
03deff1be5d0 is described below
commit 03deff1be5d09f91bcbe3bed11c31bf1a05aa94e
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 13 01:47:00 2026 +0530
CAMEL-25543: camel-dapr - Send the saveBulk and executeTransaction state
requests (#27691)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../dapr/operations/DaprStateHandler.java | 6 +-
.../component/dapr/operations/DaprStateTest.java | 96 ++++++++++++++++++++++
2 files changed, 100 insertions(+), 2 deletions(-)
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 dc4eab16111c..d9ea0370c855 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
@@ -87,7 +87,8 @@ public class DaprStateHandler implements DaprOperationHandler
{
SaveStateRequest stateRequest = new SaveStateRequest(stateStore);
stateRequest.setStates(states);
- client.saveBulkState(stateRequest);
+ // the request is only sent when the returned Mono is subscribed to
+ client.saveBulkState(stateRequest).block();
return DaprOperationResponse.create(stateRequest);
}
@@ -146,7 +147,8 @@ public class DaprStateHandler implements
DaprOperationHandler {
stateRequest.setOperations(transactions);
stateRequest.setMetadata(metadata);
- client.executeStateTransaction(stateRequest);
+ // the request is only sent when the returned Mono is subscribed to
+ client.executeStateTransaction(stateRequest).block();
return DaprOperationResponse.create(stateRequest);
}
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 58e62682139e..f2354366d799 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
@@ -17,6 +17,7 @@
package org.apache.camel.component.dapr.operations;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import io.dapr.client.DaprClient;
@@ -46,6 +47,7 @@ import reactor.core.publisher.Mono;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
@@ -134,6 +136,51 @@ public class DaprStateTest extends CamelTestSupport {
assertEquals(states, saveStateRequest.getStates());
}
+ @Test
+ void testSaveBulkSendsTheStates() {
+ // as the Dapr client, the Mono sends the request only when it is
subscribed to
+ List<State<?>> saved = new ArrayList<>();
+ when(endpoint.getClient()).thenReturn(client);
+ when(client.saveBulkState(any())).thenAnswer(invocation -> {
+ SaveStateRequest request = invocation.getArgument(0);
+ return Mono.fromRunnable(() -> saved.addAll(request.getStates()));
+ });
+
+ DaprConfiguration configuration = new DaprConfiguration();
+ configuration.setOperation(DaprOperation.state);
+ configuration.setStateOperation(StateOperation.saveBulk);
+ configuration.setStateStore("myStateStore");
+ DaprConfigurationOptionsProxy configurationOptionsProxy = new
DaprConfigurationOptionsProxy(configuration);
+
+ final Exchange exchange = new DefaultExchange(context);
+ List<State<?>> states = List.of(new State<>("k1"), new State<>("k2"));
+ exchange.getMessage().setHeader(DaprConstants.STATES, states);
+
+ new DaprStateHandler(configurationOptionsProxy,
endpoint).handle(exchange);
+
+ assertEquals(states, saved);
+ }
+
+ @Test
+ void testSaveBulkFailure() {
+ IllegalStateException failure = new IllegalStateException("the state
store rejected the request");
+ when(endpoint.getClient()).thenReturn(client);
+ when(client.saveBulkState(any())).thenReturn(Mono.error(failure));
+
+ DaprConfiguration configuration = new DaprConfiguration();
+ configuration.setOperation(DaprOperation.state);
+ configuration.setStateOperation(StateOperation.saveBulk);
+ configuration.setStateStore("myStateStore");
+ DaprConfigurationOptionsProxy configurationOptionsProxy = new
DaprConfigurationOptionsProxy(configuration);
+
+ final Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader(DaprConstants.STATES, List.of(new
State<>("k1")));
+
+ final DaprStateHandler operation = new
DaprStateHandler(configurationOptionsProxy, endpoint);
+ IllegalStateException thrown =
assertThrows(IllegalStateException.class, () -> operation.handle(exchange));
+ assertSame(failure, thrown);
+ }
+
@Test
void testSaveBulkConfiguration() {
DaprConfiguration configuration = new DaprConfiguration();
@@ -323,6 +370,55 @@ public class DaprStateTest extends CamelTestSupport {
assertEquals(transactions, executeRequest.getOperations());
}
+ @Test
+ void testExecuteTransactionSendsTheOperations() {
+ // as the Dapr client, the Mono sends the request only when it is
subscribed to
+ List<TransactionalStateOperation<?>> executed = new ArrayList<>();
+ when(endpoint.getClient()).thenReturn(client);
+
when(client.executeStateTransaction(any(ExecuteStateTransactionRequest.class))).thenAnswer(invocation
-> {
+ ExecuteStateTransactionRequest request = invocation.getArgument(0);
+ return Mono.fromRunnable(() ->
executed.addAll(request.getOperations()));
+ });
+
+ DaprConfiguration configuration = new DaprConfiguration();
+ configuration.setOperation(DaprOperation.state);
+ configuration.setStateOperation(StateOperation.executeTransaction);
+ configuration.setStateStore("myStateStore");
+ DaprConfigurationOptionsProxy configurationOptionsProxy = new
DaprConfigurationOptionsProxy(configuration);
+
+ final Exchange exchange = new DefaultExchange(context);
+ TransactionalStateOperation.OperationType op =
TransactionalStateOperation.OperationType.UPSERT;
+ List<TransactionalStateOperation<?>> transactions
+ = List.of(new TransactionalStateOperation<>(op, new
State<>("myKey")));
+ exchange.getIn().setHeader(DaprConstants.TRANSACTIONS, transactions);
+
+ new DaprStateHandler(configurationOptionsProxy,
endpoint).handle(exchange);
+
+ assertEquals(transactions, executed);
+ }
+
+ @Test
+ void testExecuteTransactionFailure() {
+ IllegalStateException failure = new IllegalStateException("the state
store rejected the transaction");
+ when(endpoint.getClient()).thenReturn(client);
+
when(client.executeStateTransaction(any(ExecuteStateTransactionRequest.class))).thenReturn(Mono.error(failure));
+
+ DaprConfiguration configuration = new DaprConfiguration();
+ configuration.setOperation(DaprOperation.state);
+ configuration.setStateOperation(StateOperation.executeTransaction);
+ configuration.setStateStore("myStateStore");
+ DaprConfigurationOptionsProxy configurationOptionsProxy = new
DaprConfigurationOptionsProxy(configuration);
+
+ final Exchange exchange = new DefaultExchange(context);
+ TransactionalStateOperation.OperationType op =
TransactionalStateOperation.OperationType.UPSERT;
+ exchange.getIn().setHeader(DaprConstants.TRANSACTIONS,
+ List.<TransactionalStateOperation<?>> of(new
TransactionalStateOperation<>(op, new State<>("myKey"))));
+
+ final DaprStateHandler operation = new
DaprStateHandler(configurationOptionsProxy, endpoint);
+ IllegalStateException thrown =
assertThrows(IllegalStateException.class, () -> operation.handle(exchange));
+ assertSame(failure, thrown);
+ }
+
@Test
void testExecuteTransactionConfiguration() {
DaprConfiguration configuration = new DaprConfiguration();
