oscerd commented on code in PR #29080: URL: https://github.com/apache/camel/pull/29080#discussion_r4454347344
########## components/camel-state-store/camel-state-store/src/test/java/org/apache/camel/component/statestore/StateStoreStreamBodyTest.java: ########## @@ -0,0 +1,150 @@ +/* + * 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.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.converter.stream.InputStreamCache; +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 }; + + @Test + void testPutStreamCacheStoresItsBytes() { + // the route caches the stream, so the producer gets a stream cache + template.requestBodyAndHeaders("direct:put", new ByteArrayInputStream(BINARY), + Map.of(StateStoreConstants.KEY, "key1")); + + assertStoredBytes("key1", BINARY); + } + + @Test + void testPutInputStreamStoresItsBytes() { + // without stream caching the producer gets the InputStream itself + template.requestBodyAndHeaders("direct:putNoStreamCaching", new ByteArrayInputStream(BINARY), + Map.of(StateStoreConstants.KEY, "key1")); + + assertStoredBytes("key1", BINARY); Review Comment: Done. Each test now uses keys of its own, such as `streamCache`, `inputStream`, `partlyRead` and `otherBytes`, so no two tests share a key. The new spooling test is in a class of its own and uses its own store. _Claude Code on behalf of Andrea Cosentino (oscerd)_ ########## components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java: ########## @@ -127,6 +132,22 @@ private long determineTtl(Exchange exchange) { 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 stream is read and stored as its bytes, which keep binary content as it is; a + * character stream (also a cached one) as its text in the charset of the exchange. + */ + private static Object valueToStore(Exchange exchange, Message message) throws InvalidPayloadException { + Object body = message.getBody(); + if (body instanceof Reader) { + return message.getMandatoryBody(String.class).getBytes(ExchangeHelper.getCharset(exchange)); + } + if (body instanceof StreamCache || body instanceof InputStream) { + return message.getMandatoryBody(byte[].class); Review Comment: The reset is in: `valueToStore` now resets a `StreamCache` before it reads it. `testPutPartlyReadStreamCacheStoresAllItsBytes` sends an `InputStreamCache` of 7 bytes, 2 of them already read, through the `noStreamCaching()` route, and all 7 bytes are stored. The previous head stored the last 5. The value now also becomes the body before the backend calls, but that does not change what a redelivery stores. Before each redelivery, the error handler copies the IN message back from the copy it took before the first attempt (`prepareExchangeForRedelivery` calls `exchange.getIn().copyFrom(original)`), so that the message is redelivered as it came and not as a step changed it. It then resets only a stream cache. On a `noStreamCaching()` route, the retry therefore gets back the plain `InputStream` that the first attempt has read, and still stores `byte[0]`. I wrote the test you describe (a backend whose first `get` fails, with `maximumRedeliveries(1)`), and it fails with the change as well. With stream caching on, which is the default, the error handler resets the cache before the redelivery and the retry stores all the bytes. `testRedeliveryStoresAllTheBytes` checks that with the same failing backend. No producer can read a plain stream that is not cached a second time; stream caching exists for that. Setting the body does change what the error handler gets after a failed attempt: a dead letter channel now receives the bytes instead of a stream that is already read, which `testFailedPutLeavesTheBytesAsTheBody` checks on the `noStreamCaching()` route. If a redelivery without stream caching should work as well, the producer could mark a stream that supports it, read it and reset it, so that the retry reads it again. I left that out, because the producer would then no longer close such a stream after reading it. _Claude Code on behalf of Andrea Cosentino (oscerd)_ ########## components/camel-state-store/camel-state-store/src/main/java/org/apache/camel/component/statestore/StateStoreProducer.java: ########## @@ -127,6 +132,22 @@ private long determineTtl(Exchange exchange) { 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 stream is read and stored as its bytes, which keep binary content as it is; a + * character stream (also a cached one) as its text in the charset of the exchange. + */ + private static Object valueToStore(Exchange exchange, Message message) throws InvalidPayloadException { + Object body = message.getBody(); + if (body instanceof Reader) { + return message.getMandatoryBody(String.class).getBytes(ExchangeHelper.getCharset(exchange)); Review Comment: Agreed. A `Reader` holds text that is already decoded, and a text body is stored as a `String`. So the text of a `Reader`, cached or not, is now stored as a `String` too, and no charset is used. `testPutReaderStoresItsText` puts `"café"` as a `StringReader` with `CamelCharsetName=ISO-8859-1`, through a route with stream caching and one without. A `get` gives `"café"`, and so does a `get` followed by `convertBodyTo(String.class)`. `testPutReaderWithACharsetNameThatJavaDoesNotKnow` does the same with `CamelCharsetName=utf8mb4`, and the put no longer fails. _Claude Code on behalf of Andrea Cosentino (oscerd)_ ########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc: ########## @@ -4742,6 +4742,15 @@ jsonata read the text as one JSON string, refused bytes and failed on a stream. the exchange when one is set; otherwise the JSON parser detects the encoding. Such text that is not valid JSON now fails with the parse error. Other text, a `Map`, a `List` and other objects are used as before. +=== camel-state-store - a stream body is stored as its bytes + +The `put` and `putIfAbsent` operations of the `state-store` component now read a stream body (an `InputStream`, a +`Reader` or the stream cache of Camel) and store its bytes, so a later `get` gives a `byte[]`. A `Reader` is stored as +its text in the charset of the exchange, UTF-8 by default. 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. Convert the `byte[]` to the type you need, +for example with `convertBodyTo(String.class)` for text. Other bodies are still stored as they are. Review Comment: Done. The upgrade guide entry and the component doc now say that the charset of the stored bytes is not stored with them. They also say that `convertBodyTo(String.class)` after a `get` decodes the bytes in the charset of that exchange, UTF-8 by default, and to use `convertBodyTo(String.class, "<charset>")` when the stored bytes are not UTF-8. As a `Reader` is now stored as text, this only concerns the bytes of an `InputStream` or a stream cache. _Claude Code on behalf of Andrea Cosentino (oscerd)_ ########## docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc: ########## @@ -4742,6 +4742,15 @@ jsonata read the text as one JSON string, refused bytes and failed on a stream. the exchange when one is set; otherwise the JSON parser detects the encoding. Such text that is not valid JSON now fails with the parse error. Other text, a `Map`, a `List` and other objects are used as before. +=== camel-state-store - a stream body is stored as its bytes Review Comment: Moved. The section now comes after "camel-jsonata - the endpoint options of an http resource" and before "camel-saxon - the xquery endpoint is producer-only", so the two camel-jsonata sections are together again. _Claude Code on behalf of Andrea Cosentino (oscerd)_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
