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 7a69b7469305 CAMEL-26239: camel-state-store - put and putIfAbsent
store a stream body as its bytes (#29080)
7a69b7469305 is described below
commit 7a69b7469305d17e60f01a3d4b5c0371f0e478db
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 18:04:22 2026 +0100
CAMEL-26239: camel-state-store - put and putIfAbsent store a stream body as
its bytes (#29080)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/catalog/components/state-store.json | 4 +-
.../camel/catalog/docs/state-store-component.adoc | 9 +
.../camel/component/statestore/state-store.json | 4 +-
.../src/main/docs/state-store-component.adoc | 9 +
.../component/statestore/StateStoreEndpoint.java | 10 +-
.../component/statestore/StateStoreProducer.java | 33 ++-
.../StateStoreSpooledStreamBodyTest.java | 75 +++++++
.../statestore/StateStoreStreamBodyTest.java | 241 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
9 files changed, 387 insertions(+), 10 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/state-store.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/state-store.json
index 807a43c79cb4..6ba0cb8d60af 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/state-store.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/state-store.json
@@ -34,8 +34,8 @@
"CamelStateStoreTtl": { "index": 2, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "Per-message TTL override in milliseconds.
Takes precedence over the endpoint ttl option.", "constantName":
"org.apache.camel.component.statestore.StateStoreConstants#TTL" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "description": "The value to store
for put and putIfAbsent, as it is: a stream is stored as the stream and shared
by the later gets, so convert it to text or bytes first. The other operations
do not use the body" },
- "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"operation", "value": "contains", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "keys", "carriers": [
"list" ], "javaType": "java.util.Set<java.lang.String>" }, { "option":
"operation", "value": "size", "carriers": [ "value" ], "javaType":
"java.lang.Integer" }, { "option": "operation", "value": "clear", "carriers": [
"null" ] } ], "description": "By the operation, wh [...]
+ "producerIn": { "carriers": [ "any" ], "description": "The value to store
for put and putIfAbsent, as it is, except a stream: an InputStream or the
stream cache of Camel is read into memory and stored as its bytes, and a Reader
as its text, a String. The other operations do not use the body" },
+ "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"operation", "value": "contains", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "keys", "carriers": [
"list" ], "javaType": "java.util.Set<java.lang.String>" }, { "option":
"operation", "value": "size", "carriers": [ "value" ], "javaType":
"java.lang.Integer" }, { "option": "operation", "value": "clear", "carriers": [
"null" ] } ], "description": "By the operation, wh [...]
},
"properties": {
"storeName": { "index": 0, "kind": "path", "displayName": "Store Name",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The name of the state
store" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/state-store-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/state-store-component.adoc
index d43032196fa4..5cdd3f1b04d6 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/state-store-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/state-store-component.adoc
@@ -76,6 +76,15 @@ The operation to perform is set via the URI option
`operation` or the header `Ca
The key is specified via the header `CamelStateStoreKey` (required for `put`,
`putIfAbsent`, `get`, `delete`, `contains`).
+`put` and `putIfAbsent` store the body as it is, except a stream, which can be
read only once. A stream body is read
+fully into memory, even by a `putIfAbsent` that finds the key already set: an
`InputStream`, or a stream cache of one,
+is stored as its bytes, so a `get` returns a `byte[]`, and a `Reader`, cached
or not, as its text, a `String`, like a
+text body.
+
+The charset of the stored bytes is not stored with them. A
`convertBodyTo(String.class)` after a `get` decodes them in
+the charset of that exchange, UTF-8 by default, so use
`convertBodyTo(String.class, "<charset>")` when the stored bytes
+are not UTF-8.
+
== TTL (Time-to-Live)
You can set a TTL (in milliseconds) for entries in two ways:
diff --git
a/components/camel-state-store/camel-state-store/src/generated/resources/META-INF/org/apache/camel/component/statestore/state-store.json
b/components/camel-state-store/camel-state-store/src/generated/resources/META-INF/org/apache/camel/component/statestore/state-store.json
index 807a43c79cb4..6ba0cb8d60af 100644
---
a/components/camel-state-store/camel-state-store/src/generated/resources/META-INF/org/apache/camel/component/statestore/state-store.json
+++
b/components/camel-state-store/camel-state-store/src/generated/resources/META-INF/org/apache/camel/component/statestore/state-store.json
@@ -34,8 +34,8 @@
"CamelStateStoreTtl": { "index": 2, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "Per-message TTL override in milliseconds.
Takes precedence over the endpoint ttl option.", "constantName":
"org.apache.camel.component.statestore.StateStoreConstants#TTL" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "description": "The value to store
for put and putIfAbsent, as it is: a stream is stored as the stream and shared
by the later gets, so convert it to text or bytes first. The other operations
do not use the body" },
- "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"operation", "value": "contains", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "keys", "carriers": [
"list" ], "javaType": "java.util.Set<java.lang.String>" }, { "option":
"operation", "value": "size", "carriers": [ "value" ], "javaType":
"java.lang.Integer" }, { "option": "operation", "value": "clear", "carriers": [
"null" ] } ], "description": "By the operation, wh [...]
+ "producerIn": { "carriers": [ "any" ], "description": "The value to store
for put and putIfAbsent, as it is, except a stream: an InputStream or the
stream cache of Camel is read into memory and stored as its bytes, and a Reader
as its text, a String. The other operations do not use the body" },
+ "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"operation", "value": "contains", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "keys", "carriers": [
"list" ], "javaType": "java.util.Set<java.lang.String>" }, { "option":
"operation", "value": "size", "carriers": [ "value" ], "javaType":
"java.lang.Integer" }, { "option": "operation", "value": "clear", "carriers": [
"null" ] } ], "description": "By the operation, wh [...]
},
"properties": {
"storeName": { "index": 0, "kind": "path", "displayName": "Store Name",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The name of the state
store" },
diff --git
a/components/camel-state-store/camel-state-store/src/main/docs/state-store-component.adoc
b/components/camel-state-store/camel-state-store/src/main/docs/state-store-component.adoc
index d43032196fa4..5cdd3f1b04d6 100644
---
a/components/camel-state-store/camel-state-store/src/main/docs/state-store-component.adoc
+++
b/components/camel-state-store/camel-state-store/src/main/docs/state-store-component.adoc
@@ -76,6 +76,15 @@ The operation to perform is set via the URI option
`operation` or the header `Ca
The key is specified via the header `CamelStateStoreKey` (required for `put`,
`putIfAbsent`, `get`, `delete`, `contains`).
+`put` and `putIfAbsent` store the body as it is, except a stream, which can be
read only once. A stream body is read
+fully into memory, even by a `putIfAbsent` that finds the key already set: an
`InputStream`, or a stream cache of one,
+is stored as its bytes, so a `get` returns a `byte[]`, and a `Reader`, cached
or not, as its text, a `String`, like a
+text body.
+
+The charset of the stored bytes is not stored with them. A
`convertBodyTo(String.class)` after a `get` decodes them in
+the charset of that exchange, UTF-8 by default, so use
`convertBodyTo(String.class, "<charset>")` when the stored bytes
+are not UTF-8.
+
== TTL (Time-to-Live)
You can set a TTL (in milliseconds) for entries in two ways:
diff --git
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreEndpoint.java
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreEndpoint.java
index 90a855cce895..afedd7cc04c9 100644
---
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreEndpoint.java
+++
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreEndpoint.java
@@ -40,9 +40,10 @@ import org.apache.camel.support.DefaultEndpoint;
category = { Category.CACHE },
headersClass = StateStoreConstants.class)
@BodyTypes(producerIn = @BodyType(value = Carrier.ANY,
- description = "The value to store for put
and putIfAbsent, as it is: a stream is stored as"
- + " the stream and shared by
the later gets, so convert it to text or bytes"
- + " first. The other
operations do not use the body"),
+ description = "The value to store for put
and putIfAbsent, as it is, except a stream: an"
+ + " InputStream or the stream
cache of Camel is read into memory and stored as"
+ + " its bytes, and a Reader as
its text, a String. The other operations do not"
+ + " use the body"),
producerOut = @BodyType(value = Carrier.UNKNOWN,
byOption = {
@BodyType.When(option =
"operation", value = "contains", carriers = Carrier.VALUE,
@@ -57,7 +58,8 @@ import org.apache.camel.support.DefaultEndpoint;
+ " for putIfAbsent and the
removed value for delete, each no body when there"
+ " is none; a Boolean for
contains, the Set of the keys for keys, the count"
+ " for size and no body for
clear. The value is the object the backend gives"
- + " back: the in-memory one
the object stored, another may give a copy"))
+ + " back: the in-memory one
the object stored, another may give a copy; a"
+ + " stream was stored as a
byte array, and a Reader as a String"))
public class StateStoreEndpoint extends DefaultEndpoint {
@UriPath(description = "The name of the state store")
diff --git
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
index 6bdd373ea584..d76f20b7c71b 100644
---
a/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
+++
b/components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java
@@ -16,11 +16,15 @@
*/
package org.apache.camel.component.statestore;
+import java.io.InputStream;
+import java.io.Reader;
import java.time.Duration;
import java.util.Arrays;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
+import org.apache.camel.StreamCache;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.support.DefaultProducer;
@@ -54,7 +58,7 @@ public class StateStoreProducer extends DefaultProducer {
switch (op) {
case put -> {
String key = requireKey(message);
- Object value = message.getBody();
+ Object value = valueToStore(message);
// KeyValueRepository.put() is void; retrieve old value first
for backward compat
Object previous = backend.get(key);
backend.put(key, value, ttl);
@@ -62,7 +66,7 @@ public class StateStoreProducer extends DefaultProducer {
}
case putIfAbsent -> {
String key = requireKey(message);
- Object value = message.getBody();
+ Object value = valueToStore(message);
Object existing = backend.putIfAbsent(key, value, ttl);
message.setBody(existing);
}
@@ -127,6 +131,31 @@ public class StateStoreProducer extends DefaultProducer {
return endpoint.getTtl();
}
+ /**
+ * The value to store: the body as it is, except a stream, which can be
read only once and would be shared by every
+ * exchange that gets the key. A character stream (also a cached one) is
stored as its text, like a text body, and
+ * another stream as its bytes, which keep binary content as it is. The
value becomes the body as well, so that when
+ * the backend fails, the error handler (a dead letter channel, for
example) gets the value, and not the stream that
+ * is already read.
+ */
+ private static Object valueToStore(Message message) throws
InvalidPayloadException {
+ Object body = message.getBody();
+ if (body instanceof StreamCache cache) {
+ // an earlier step may have read part of it, and nothing resets it
on a route without stream caching
+ cache.reset();
+ }
+ Object value;
+ if (body instanceof Reader) {
+ value = message.getMandatoryBody(String.class);
+ } else if (body instanceof StreamCache || body instanceof InputStream)
{
+ value = message.getMandatoryBody(byte[].class);
+ } else {
+ return body;
+ }
+ message.setBody(value);
+ return value;
+ }
+
private String requireKey(Message message) {
String key = message.getHeader(StateStoreConstants.KEY, String.class);
if (key == null) {
diff --git
a/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreSpooledStreamBodyTest.java
b/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreSpooledStreamBodyTest.java
new file mode 100644
index 000000000000..e2bbd88ab533
--- /dev/null
+++
b/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreSpooledStreamBodyTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.statestore;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.converter.stream.FileInputStreamCache;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class StateStoreSpooledStreamBodyTest extends CamelTestSupport {
+
+ private static final byte[] CONTENT = "a stream that is spooled to a
file".getBytes(StandardCharsets.UTF_8);
+
+ // the body that the producer of the put gets
+ private volatile Object bodyToStore;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // spool every stream of more than one byte to a file, which is
deleted when its exchange completes
+ context.getStreamCachingStrategy().setSpoolEnabled(true);
+ context.getStreamCachingStrategy().setSpoolThreshold(1);
+ return context;
+ }
+
+ @Test
+ void testPutSpooledStreamStoresItsBytes() {
+ // stream caching keeps a ByteArrayInputStream in memory whatever its
size, so it is wrapped
+ template.requestBodyAndHeaders("direct:put", new
BufferedInputStream(new ByteArrayInputStream(CONTENT)),
+ Map.of(StateStoreConstants.KEY, "spooled"));
+ assertThat(bodyToStore).isInstanceOf(FileInputStreamCache.class);
+
+ // the file is gone now that the put has completed, but the bytes were
stored
+ for (int i = 0; i < 2; i++) {
+ Object value = template.requestBodyAndHeaders("direct:get", null,
Map.of(StateStoreConstants.KEY, "spooled"));
+ assertThat(value).isInstanceOf(byte[].class);
+ assertThat((byte[]) value).containsExactly(CONTENT);
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:put").streamCaching()
+ .process(exchange -> bodyToStore =
exchange.getMessage().getBody())
+ .to("state-store:spooledStore?operation=put");
+
from("direct:get").to("state-store:spooledStore?operation=get");
+ }
+ };
+ }
+}
diff --git
a/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreStreamBodyTest.java
b/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreStreamBodyTest.java
new file mode 100644
index 000000000000..69e1820fc302
--- /dev/null
+++
b/components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreStreamBodyTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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.statestore;
+
+import java.io.ByteArrayInputStream;
+import java.io.StringReader;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.converter.stream.InputStreamCache;
+import org.apache.camel.support.MemoryKeyValueRepository;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class StateStoreStreamBodyTest extends CamelTestSupport {
+
+ // not valid UTF-8, so it would not survive a conversion to text
+ private static final byte[] BINARY = { 0, 1, 2, (byte) 0xC3, (byte) 0x28,
(byte) 0xFF, (byte) 0xFE };
+
+ // not ASCII, so a wrong charset shows
+ private static final String TEXT = "caf\u00e9";
+
+ // the backends of the stores whose first get fails
+ private final FailingOnceBackend redeliveryBackend = new
FailingOnceBackend();
+ private final FailingOnceBackend failingBackend = new FailingOnceBackend();
+
+ @Test
+ void testPutStreamCacheStoresItsBytes() {
+ // the route caches the stream, so the producer gets a stream cache
+ put("direct:put", new ByteArrayInputStream(BINARY), "streamCache");
+
+ assertStoredBytes("streamCache", BINARY);
+ }
+
+ @Test
+ void testPutInputStreamStoresItsBytes() {
+ // without stream caching the producer gets the InputStream itself
+ put("direct:putNoStreamCaching", new ByteArrayInputStream(BINARY),
"inputStream");
+
+ assertStoredBytes("inputStream", BINARY);
+ }
+
+ @Test
+ void testPutGivenStreamCacheStoresItsBytes() {
+ put("direct:putNoStreamCaching", new InputStreamCache(BINARY),
"givenStreamCache");
+
+ assertStoredBytes("givenStreamCache", BINARY);
+ }
+
+ @Test
+ void testPutPartlyReadStreamCacheStoresAllItsBytes() {
+ InputStreamCache cache = new InputStreamCache(BINARY);
+ // an earlier step read part of it, and a route without stream caching
does not reset it
+ cache.read();
+ cache.read();
+
+ put("direct:putNoStreamCaching", cache, "partlyRead");
+
+ assertStoredBytes("partlyRead", BINARY);
+ }
+
+ @Test
+ void testRedeliveryStoresAllTheBytes() {
+ // the first get of the backend fails, and the error handler sends the
message as it came to the producer
+ // again, with the stream cache reset
+ put("direct:putRedelivery", new ByteArrayInputStream(BINARY),
"redelivered");
+
+ Object value = redeliveryBackend.get("redelivered");
+ assertThat(value).isInstanceOf(byte[].class);
+ assertThat((byte[]) value).containsExactly(BINARY);
+ }
+
+ @Test
+ void testFailedPutLeavesTheBytesAsTheBody() throws Exception {
+ MockEndpoint failed = getMockEndpoint("mock:failed");
+ failed.expectedMessageCount(1);
+
+ // the first get of the backend fails, after the put has read the
stream, which cannot be read again
+ put("direct:putFailing", new ByteArrayInputStream(BINARY), "failed");
+
+ failed.assertIsSatisfied();
+ Object body =
failed.getReceivedExchanges().get(0).getMessage().getBody();
+ assertThat(body).isInstanceOf(byte[].class);
+ assertThat((byte[]) body).containsExactly(BINARY);
+ }
+
+ @Test
+ void testPutReaderStoresItsText() {
+ // a Reader holds text, so the charset of the exchange plays no part
+ template.requestBodyAndHeaders("direct:put", new StringReader(TEXT),
+ Map.of(StateStoreConstants.KEY, "cachedReader",
Exchange.CHARSET_NAME, "ISO-8859-1"));
+ template.requestBodyAndHeaders("direct:putNoStreamCaching", new
StringReader(TEXT),
+ Map.of(StateStoreConstants.KEY, "plainReader",
Exchange.CHARSET_NAME, "ISO-8859-1"));
+
+ assertStoredText("cachedReader", TEXT);
+ assertStoredText("plainReader", TEXT);
+ }
+
+ @Test
+ void testPutReaderWithACharsetNameThatJavaDoesNotKnow() {
+ template.requestBodyAndHeaders("direct:put", new StringReader(TEXT),
+ Map.of(StateStoreConstants.KEY, "cachedReaderUnknownCharset",
Exchange.CHARSET_NAME, "utf8mb4"));
+ template.requestBodyAndHeaders("direct:putNoStreamCaching", new
StringReader(TEXT),
+ Map.of(StateStoreConstants.KEY, "plainReaderUnknownCharset",
Exchange.CHARSET_NAME, "utf8mb4"));
+
+ assertStoredText("cachedReaderUnknownCharset", TEXT);
+ assertStoredText("plainReaderUnknownCharset", TEXT);
+ }
+
+ @Test
+ void testPutIfAbsentReaderStoresItsText() {
+ put("direct:putIfAbsentNoStreamCaching", new StringReader("hello
world"), "putIfAbsentReader");
+
+ assertStoredText("putIfAbsentReader", "hello world");
+ }
+
+ @Test
+ void testPutIfAbsentStreamStoresItsBytes() {
+ Object existing = template.requestBodyAndHeaders("direct:putIfAbsent",
new ByteArrayInputStream(BINARY),
+ Map.of(StateStoreConstants.KEY, "putIfAbsentStream"));
+ assertThat(existing).isNull();
+
+ assertStoredBytes("putIfAbsentStream", BINARY);
+ }
+
+ @Test
+ void testPutGivesThePreviousBytes() {
+ put("direct:put", new ByteArrayInputStream(BINARY), "previous");
+
+ Object previous = template.requestBodyAndHeaders("direct:put", "next",
+ Map.of(StateStoreConstants.KEY, "previous"));
+ assertThat(previous).isInstanceOf(byte[].class);
+ assertThat((byte[]) previous).containsExactly(BINARY);
+ }
+
+ @Test
+ void testOtherBodiesAreStoredAsTheyAre() {
+ byte[] bytes = { 1, 2, 3 };
+ List<String> list = List.of("a", "b");
+ Map<String, Object> map = Map.of("name", "camel");
+
+ put("direct:put", bytes, "otherBytes");
+ put("direct:put", list, "otherList");
+ put("direct:put", map, "otherMap");
+ put("direct:put", "text", "otherText");
+
+ assertThat(get("otherBytes")).isSameAs(bytes);
+ assertThat(get("otherList")).isSameAs(list);
+ assertThat(get("otherMap")).isSameAs(map);
+ assertThat(get("otherText")).isEqualTo("text");
+ }
+
+ private void put(String uri, Object body, String key) {
+ template.requestBodyAndHeaders(uri, body,
Map.of(StateStoreConstants.KEY, key));
+ }
+
+ private void assertStoredBytes(String key, byte[] expected) {
+ // every get gives the whole content, not a stream that an earlier get
has read
+ for (int i = 0; i < 3; i++) {
+ Object value = get(key);
+ assertThat(value).isInstanceOf(byte[].class);
+ assertThat((byte[]) value).containsExactly(expected);
+ }
+ }
+
+ private void assertStoredText(String key, String expected) {
+ assertThat(get(key)).isEqualTo(expected);
+ // and as a route that gets it as text reads it
+ assertThat(template.requestBodyAndHeaders("direct:getText", null,
Map.of(StateStoreConstants.KEY, key)))
+ .isEqualTo(expected);
+ }
+
+ private Object get(String key) {
+ return template.requestBodyAndHeaders("direct:get", null,
Map.of(StateStoreConstants.KEY, key));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ StateStoreEndpoint redelivery
+ =
getContext().getEndpoint("state-store:redeliveryStore?operation=put",
StateStoreEndpoint.class);
+ redelivery.setBackend(redeliveryBackend);
+ StateStoreEndpoint failing
+ =
getContext().getEndpoint("state-store:failingStore?operation=put",
StateStoreEndpoint.class);
+ failing.setBackend(failingBackend);
+
+
from("direct:put").streamCaching().to("state-store:myStore?operation=put");
+
from("direct:putIfAbsent").streamCaching().to("state-store:myStore?operation=putIfAbsent");
+
from("direct:putNoStreamCaching").noStreamCaching().to("state-store:myStore?operation=put");
+ from("direct:putIfAbsentNoStreamCaching").noStreamCaching()
+ .to("state-store:myStore?operation=putIfAbsent");
+ from("direct:putRedelivery").streamCaching()
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .to(redelivery);
+ from("direct:putFailing").noStreamCaching()
+ .errorHandler(deadLetterChannel("mock:failed"))
+ .to(failing);
+ from("direct:get").to("state-store:myStore?operation=get");
+
from("direct:getText").to("state-store:myStore?operation=get").convertBodyTo(String.class);
+ }
+ };
+ }
+
+ /**
+ * A backend whose first get fails, as a remote backend may on a transient
error.
+ */
+ private static final class FailingOnceBackend extends
MemoryKeyValueRepository {
+
+ private final AtomicBoolean failed = new AtomicBoolean();
+
+ @Override
+ public Object get(String key) {
+ if (failed.compareAndSet(false, true)) {
+ throw new IllegalStateException("The first get of the backend
fails");
+ }
+ return super.get(key);
+ }
+ }
+}
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 cef13898d65b..f36c5a8746de 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
@@ -5262,6 +5262,18 @@ For a `jsonata:http:` or `jsonata:https:` resource, the
options of the endpoint,
`prettyPrint`, are now used by the endpoint. Previously they were appended to
the URL of the resource and sent to the
server, and the endpoint used its defaults. Only the other query parameters
are sent to the server.
+=== camel-state-store - a stream body is stored as its bytes, and a Reader as
its text
+
+The `put` and `putIfAbsent` operations of the `state-store` component now read
a stream body fully into memory and
+store its content: an `InputStream` or the stream cache of Camel as its bytes,
so a later `get` gives a `byte[]`, and a
+`Reader` as its text, a `String`, like a text body. Before, the stream object
itself was stored: the first `get` could
+read it, later ones got an empty stream, and every exchange that got the key
shared the same stream object. A backend
+that serializes its values could not store such a stream at all. Other bodies
are still stored as they are.
+
+The charset of the stored bytes is not stored with them. A
`convertBodyTo(String.class)` after a `get` decodes them in
+the charset of that exchange, UTF-8 by default, so use
`convertBodyTo(String.class, "<charset>")` when the stored bytes
+are not UTF-8.
+
=== camel-saxon - the xquery endpoint is producer-only
The `xquery` endpoint can only be used as a producer. A route that starts
`from("xquery:...")` now fails when it
