You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2022/05/10 13:40:34 UTC
[camel] branch stream-caching created (now 42c45981a83)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch stream-caching
in repository https://gitbox.apache.org/repos/asf/camel.git
at 42c45981a83 CAMEL-18087: camel-core - Enable stream caching by default.
This branch includes the following new commits:
new 42c45981a83 CAMEL-18087: camel-core - Enable stream caching by default.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[camel] 01/01: CAMEL-18087: camel-core - Enable stream caching by default.
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch stream-caching
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 42c45981a83a7abd6ef5a9376c26dfc77e0cf3ac
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue May 10 15:39:59 2022 +0200
CAMEL-18087: camel-core - Enable stream caching by default.
---
.../main/camel-main-configuration-metadata.json | 2 +-
.../camel/component/jetty/HttpConverterTest.java | 1 +
.../camel/spring/DefaultStreamCachingTest.java | 4 +--
.../org/apache/camel/RuntimeConfiguration.java | 2 +-
.../camel/impl/engine/AbstractCamelContext.java | 7 +++--
.../camel/impl/engine/CamelInternalProcessor.java | 25 +++++++++++------
.../impl/engine/DefaultStreamCachingStrategy.java | 7 ++++-
.../impl/converter/CoreTypeConverterRegistry.java | 12 +++++---
.../component/file/GenericFileConverterTest.java | 32 +++++++++++++++++++++-
.../apache/camel/impl/LogDebugBodyStreamsTest.java | 8 ++++++
...istryStatisticsEnabledNoStreamCachingTest.java} | 7 +++--
...TypeConverterRegistryStatisticsEnabledTest.java | 4 +--
.../processor/StreamCacheInternalErrorTest.java | 5 ++++
.../processor/interceptor/NoStreamCachingTest.java | 9 +++---
.../transformer/TransformerContractTest.java | 2 +-
.../camel-main-configuration-metadata.json | 2 +-
core/camel-main/src/main/docs/main.adoc | 2 +-
.../camel/main/DefaultConfigurationProperties.java | 12 ++++++--
.../management/ManagedNonManagedServiceTest.java | 2 +-
...edProducerRouteAddRemoveRegisterAlwaysTest.java | 2 +-
.../management/ManagedRouteAddRemoveTest.java | 2 +-
.../ManagedTypeConverterRegistryTest.java | 4 +--
.../org/apache/camel/support/MessageHelper.java | 3 +-
.../ROOT/pages/camel-3x-upgrade-guide-3_17.adoc | 31 +++++++++++++++++++++
24 files changed, 147 insertions(+), 40 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
index 4150355c981..668d2cf6ae3 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
@@ -108,7 +108,7 @@
{ "name": "camel.main.startupSummaryLevel", "description": "Controls the level of information logged during startup (and shutdown) of CamelContext.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "object", "javaType": "org.apache.camel.StartupSummaryLevel", "defaultValue": "Default" },
{ "name": "camel.main.streamCachingAnySpoolRules", "description": "Sets whether if just any of the org.apache.camel.spi.StreamCachingStrategy.SpoolRule rules returns true then shouldSpoolCache(long) returns true, to allow spooling to disk. If this option is false, then all the org.apache.camel.spi.StreamCachingStrategy.SpoolRule must return true. The default value is false which means that all the rules must return true.", "sourceType": "org.apache.camel.main.DefaultConfigurationProp [...]
{ "name": "camel.main.streamCachingBufferSize", "description": "Sets the stream caching buffer size to use when allocating in-memory buffers used for in-memory stream caches. The default size is 4096.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "int" },
- { "name": "camel.main.streamCachingEnabled", "description": "Sets whether stream caching is enabled or not. Default is false.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": "false" },
+ { "name": "camel.main.streamCachingEnabled", "description": "Sets whether stream caching is enabled or not. While stream types (like StreamSource, InputStream and Reader) are commonly used in messaging for performance reasons, they also have an important drawback: they can only be read once. In order to be able to work with message content multiple times, the stream needs to be cached. Streams are cached in memory. However, for large stream messages (over 128 KB by default) will be c [...]
{ "name": "camel.main.streamCachingRemoveSpoolDirectoryWhenStopping", "description": "Whether to remove stream caching temporary directory when stopping. This option is default true.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": true },
{ "name": "camel.main.streamCachingSpoolCipher", "description": "Sets a stream caching cipher name to use when spooling to disk to write with encryption. By default the data is not encrypted.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
{ "name": "camel.main.streamCachingSpoolDirectory", "description": "Sets the stream caching spool (temporary) directory to use for overflow and spooling to disk. If no spool directory has been explicit configured, then a temporary directory is created in the java.io.tmpdir directory.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
diff --git a/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/HttpConverterTest.java b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/HttpConverterTest.java
index dbab66205af..4552de5116b 100644
--- a/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/HttpConverterTest.java
+++ b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/HttpConverterTest.java
@@ -112,6 +112,7 @@ public class HttpConverterTest extends BaseJettyTest {
}).transform(constant("Bye World"));
}
});
+ context.setStreamCaching(false); // this test requires stream caching disabled to work with raw servlet
context.start();
String out = template.requestBody("http://localhost:{{port}}/test", "Hello World", String.class);
diff --git a/components/camel-spring-xml/src/test/java/org/apache/camel/spring/DefaultStreamCachingTest.java b/components/camel-spring-xml/src/test/java/org/apache/camel/spring/DefaultStreamCachingTest.java
index 1371fd42575..3d9672c41f0 100644
--- a/components/camel-spring-xml/src/test/java/org/apache/camel/spring/DefaultStreamCachingTest.java
+++ b/components/camel-spring-xml/src/test/java/org/apache/camel/spring/DefaultStreamCachingTest.java
@@ -22,7 +22,7 @@ import org.junit.jupiter.api.Test;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class DefaultStreamCachingTest {
@@ -31,7 +31,7 @@ public class DefaultStreamCachingTest {
AbstractApplicationContext appContext
= new ClassPathXmlApplicationContext(new String[] { "org/apache/camel/spring/streamCaching.xml" });
CamelContext camelContext = appContext.getBean("camelContext", CamelContext.class);
- assertFalse(camelContext.isStreamCaching(), "StreamCaching should not be enabled");
+ assertTrue(camelContext.isStreamCaching(), "StreamCaching should be enabled by default");
// we're done so let's properly close the application context
IOHelper.close(appContext);
diff --git a/core/camel-api/src/main/java/org/apache/camel/RuntimeConfiguration.java b/core/camel-api/src/main/java/org/apache/camel/RuntimeConfiguration.java
index 30317b395de..2e08ffe5b5b 100644
--- a/core/camel-api/src/main/java/org/apache/camel/RuntimeConfiguration.java
+++ b/core/camel-api/src/main/java/org/apache/camel/RuntimeConfiguration.java
@@ -23,7 +23,7 @@ package org.apache.camel;
public interface RuntimeConfiguration {
/**
- * Sets whether stream caching is enabled or not (default is disabled).
+ * Sets whether stream caching is enabled or not (default is enabled).
*
* @param cache whether stream caching is enabled or not
*/
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index 0134deb1549..88b2c2e81e9 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -265,7 +265,7 @@ public abstract class AbstractCamelContext extends BaseService
private Boolean messageHistory = Boolean.FALSE;
private Boolean logMask = Boolean.FALSE;
private Boolean logExhaustedMessageBody = Boolean.FALSE;
- private Boolean streamCache = Boolean.FALSE;
+ private Boolean streamCache = Boolean.TRUE;
private Boolean disableJMX = Boolean.FALSE;
private Boolean loadTypeConverters = Boolean.FALSE;
private Boolean loadHealthChecks = Boolean.FALSE;
@@ -3209,8 +3209,9 @@ public abstract class AbstractCamelContext extends BaseService
getClassResolver(),
getPackageScanClassResolver(), getApplicationContextClassLoader(), getRouteController());
}
- if (isStreamCaching()) {
- LOG.info("StreamCaching is enabled on CamelContext: {}", getName());
+ if (!isStreamCaching()) {
+ // stream caching is default enabled so lets report if it has been disabled
+ LOG.info("StreamCaching is disabled on CamelContext: {}", getName());
}
if (isBacklogTracing()) {
// tracing is added in the DefaultChannel so we can enable it on the fly
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
index d723ee1e221..031bec38fdb 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
@@ -913,14 +913,23 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor implements In
@Override
public StreamCache before(Exchange exchange) throws Exception {
// check if body is already cached
- Object body = exchange.getIn().getBody();
- if (body == null) {
- return null;
- } else if (body instanceof StreamCache) {
- StreamCache sc = (StreamCache) body;
- // reset so the cache is ready to be used before processing
- sc.reset();
- return sc;
+ try {
+ Object body = exchange.getIn().getBody();
+ if (body == null) {
+ return null;
+ } else if (body instanceof StreamCache) {
+ StreamCache sc = (StreamCache) body;
+ // reset so the cache is ready to be used before processing
+ sc.reset();
+ return sc;
+ }
+ } catch (Exception e) {
+ // lets allow Camels error handler to deal with stream cache failures
+ StreamCacheException tce = new StreamCacheException(null, e);
+ exchange.setException(tce);
+ // because this is stream caching error then we cannot use redelivery as the message body is corrupt
+ // so mark as redelivery exhausted
+ exchange.adapt(ExtendedExchange.class).setRedeliveryExhausted(true);
}
// check if we somewhere failed due to a stream caching exception
Throwable cause = exchange.getException();
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
index 6c6733e1b7b..2d0bd1ca242 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
@@ -204,7 +204,12 @@ public class DefaultStreamCachingStrategy extends ServiceSupport implements Came
@Override
public StreamCache cache(Exchange exchange) {
Message message = exchange.getMessage();
- StreamCache cache = message.getBody(StreamCache.class);
+ StreamCache cache = null;
+ // try convert to stream cache
+ Object body = message.getBody();
+ if (body != null) {
+ cache = camelContext.getTypeConverter().convertTo(StreamCache.class, exchange, body);
+ }
if (cache != null) {
if (LOG.isTraceEnabled()) {
LOG.trace("Cached stream to {} -> {}", cache.inMemory() ? "memory" : "spool", cache);
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/converter/CoreTypeConverterRegistry.java b/core/camel-base/src/main/java/org/apache/camel/impl/converter/CoreTypeConverterRegistry.java
index 14f301c63ce..5983cd80b5b 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/converter/CoreTypeConverterRegistry.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/converter/CoreTypeConverterRegistry.java
@@ -350,11 +350,15 @@ public class CoreTypeConverterRegistry extends ServiceSupport implements TypeCon
protected Object doConvertTo(
final Class<?> type, final Exchange exchange, final Object value,
final boolean mandatory, final boolean tryConvert) {
+
+ boolean statisticsEnabled = !tryConvert && statistics.isStatisticsEnabled(); // we only capture if not try-convert in use
+
Object answer;
try {
answer = doConvertTo(type, exchange, value, tryConvert);
} catch (Exception e) {
- if (statistics.isStatisticsEnabled()) {
+ // only record if not try
+ if (statisticsEnabled) {
failedCounter.increment();
}
if (tryConvert) {
@@ -373,12 +377,12 @@ public class CoreTypeConverterRegistry extends ServiceSupport implements TypeCon
}
if (answer == TypeConverter.MISS_VALUE) {
// Could not find suitable conversion
- if (statistics.isStatisticsEnabled()) {
+ if (statisticsEnabled) {
missCounter.increment();
}
return null;
} else {
- if (statistics.isStatisticsEnabled()) {
+ if (statisticsEnabled) {
hitCounter.increment();
}
return answer;
@@ -390,7 +394,7 @@ public class CoreTypeConverterRegistry extends ServiceSupport implements TypeCon
final boolean tryConvert)
throws Exception {
boolean trace = LOG.isTraceEnabled();
- boolean statisticsEnabled = statistics.isStatisticsEnabled();
+ boolean statisticsEnabled = !tryConvert && statistics.isStatisticsEnabled(); // we only capture if not try-convert in use
if (trace) {
LOG.trace("Finding type converter to convert {} -> {} with value: {}",
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/GenericFileConverterTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/GenericFileConverterTest.java
index 1b607c4594a..3ca41cb057a 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/file/GenericFileConverterTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/GenericFileConverterTest.java
@@ -26,6 +26,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.converter.stream.InputStreamCache;
import org.junit.jupiter.api.Test;
public class GenericFileConverterTest extends ContextTestSupport {
@@ -143,13 +144,42 @@ public class GenericFileConverterTest extends ContextTestSupport {
@Override
public void process(Exchange exchange) throws Exception {
Object body = exchange.getIn().getBody();
- assertIsInstanceOf(BufferedInputStream.class, body);
+ assertIsInstanceOf(InputStreamCache.class, body);
}
}).to("mock:result");
}
});
context.start();
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.message(0).body().isInstanceOf(InputStreamCache.class);
+ mock.message(0).body(String.class).isEqualTo("Hello World");
+
+ template.sendBodyAndHeader(fileUri(), "Hello World", Exchange.FILE_NAME, "hello.txt");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testToFileInputStreamNoStreamCaching() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from(fileUri("?initialDelay=0&delay=10"))
+ .noStreamCaching()
+ .convertBodyTo(InputStream.class).process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ Object body = exchange.getIn().getBody();
+ assertIsInstanceOf(BufferedInputStream.class, body);
+ }
+ }).to("mock:result");
+ }
+ });
+ context.start();
+
// a file input stream is wrapped in a buffered so its faster
MockEndpoint mock = getMockEndpoint("mock:result");
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/LogDebugBodyStreamsTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/LogDebugBodyStreamsTest.java
index cb9b4a2df1d..e6bea79458e 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/LogDebugBodyStreamsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/LogDebugBodyStreamsTest.java
@@ -19,6 +19,7 @@ package org.apache.camel.impl;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
+import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
@@ -31,6 +32,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class LogDebugBodyStreamsTest extends ContextTestSupport {
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.setStreamCaching(false); // turn off stream caching as this test case requires that
+ return context;
+ }
+
@Override
protected Registry createRegistry() throws Exception {
Registry registry = super.createRegistry();
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledNoStreamCachingTest.java
similarity index 90%
copy from core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java
copy to core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledNoStreamCachingTest.java
index 7e5a95d0af6..616fcab5ca9 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledNoStreamCachingTest.java
@@ -22,13 +22,16 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spi.TypeConverterRegistry;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
-public class TypeConverterRegistryStatisticsEnabledTest extends ContextTestSupport {
+public class TypeConverterRegistryStatisticsEnabledNoStreamCachingTest extends ContextTestSupport {
@Override
protected CamelContext createCamelContext() throws Exception {
CamelContext context = super.createCamelContext();
+ context.setStreamCaching(false);
context.setTypeConverterStatisticsEnabled(true);
return context;
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java
index 7e5a95d0af6..ab6fb661a76 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/TypeConverterRegistryStatisticsEnabledTest.java
@@ -48,7 +48,7 @@ public class TypeConverterRegistryStatisticsEnabledTest extends ContextTestSuppo
Long failed = reg.getStatistics().getFailedCounter();
assertEquals(0, failed.intValue());
Long miss = reg.getStatistics().getMissCounter();
- assertEquals(0, miss.intValue());
+ assertEquals(4, miss.intValue()); // stream caching misses
try {
template.sendBody("direct:start", "foo");
@@ -61,7 +61,7 @@ public class TypeConverterRegistryStatisticsEnabledTest extends ContextTestSuppo
failed = reg.getStatistics().getFailedCounter();
assertEquals(1, failed.intValue());
miss = reg.getStatistics().getMissCounter();
- assertEquals(0, miss.intValue());
+ assertEquals(5, miss.intValue()); // stream caching misses
// reset
reg.getStatistics().reset();
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/StreamCacheInternalErrorTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/StreamCacheInternalErrorTest.java
index ec84f24c684..f49e265865a 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/StreamCacheInternalErrorTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/StreamCacheInternalErrorTest.java
@@ -105,6 +105,11 @@ public class StreamCacheInternalErrorTest extends ContextTestSupport {
@Override
public <T> T convertTo(Class<T> type, Exchange exchange, Object value) throws TypeConversionException {
+ return tryConvertTo(type, exchange, value);
+ }
+
+ @Override
+ public <T> T tryConvertTo(Class<T> type, Exchange exchange, Object value) throws TypeConversionException {
invoked++;
String str = value.toString();
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/NoStreamCachingTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/NoStreamCachingTest.java
index f0a0790e8c7..6a8d7097d61 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/NoStreamCachingTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/NoStreamCachingTest.java
@@ -70,7 +70,7 @@ public class NoStreamCachingTest extends ContextTestSupport {
}
@Test
- public void testNoStreamCacheIsDefault() throws Exception {
+ public void testStreamCacheIsDefault() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -85,8 +85,9 @@ public class NoStreamCachingTest extends ContextTestSupport {
template.sendBody("direct:a", message);
assertMockEndpointsSatisfied();
- boolean b1 = a.assertExchangeReceived(0).getIn().getBody() instanceof ByteArrayInputStream;
- assertTrue(b1);
+
+ boolean a1 = a.assertExchangeReceived(0).getIn().getBody() instanceof StreamCache;
+ assertTrue(a1);
assertEquals(MESSAGE, a.assertExchangeReceived(0).getIn().getBody(String.class));
}
@@ -95,7 +96,7 @@ public class NoStreamCachingTest extends ContextTestSupport {
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:a").to("mock:a");
+ from("direct:a").noStreamCaching().to("mock:a");
from("direct:b").streamCaching().to("mock:b");
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/transformer/TransformerContractTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/transformer/TransformerContractTest.java
index adae740b6d8..7d177581eb8 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/transformer/TransformerContractTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/transformer/TransformerContractTest.java
@@ -117,7 +117,7 @@ public class TransformerContractTest extends ContextTestSupport {
assertEquals("<foo/>", exa.getIn().getBody());
assertEquals(A.class, exb.getIn().getBody().getClass());
assertEquals(B.class, exa2.getIn().getBody().getClass());
- assertEquals("<fooResponse/>", new String((byte[]) answer.getIn().getBody()));
+ assertEquals("<fooResponse/>", answer.getIn().getBody(String.class));
}
public static class MyTypeConverters implements TypeConverters {
diff --git a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index 4150355c981..668d2cf6ae3 100644
--- a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++ b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -108,7 +108,7 @@
{ "name": "camel.main.startupSummaryLevel", "description": "Controls the level of information logged during startup (and shutdown) of CamelContext.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "object", "javaType": "org.apache.camel.StartupSummaryLevel", "defaultValue": "Default" },
{ "name": "camel.main.streamCachingAnySpoolRules", "description": "Sets whether if just any of the org.apache.camel.spi.StreamCachingStrategy.SpoolRule rules returns true then shouldSpoolCache(long) returns true, to allow spooling to disk. If this option is false, then all the org.apache.camel.spi.StreamCachingStrategy.SpoolRule must return true. The default value is false which means that all the rules must return true.", "sourceType": "org.apache.camel.main.DefaultConfigurationProp [...]
{ "name": "camel.main.streamCachingBufferSize", "description": "Sets the stream caching buffer size to use when allocating in-memory buffers used for in-memory stream caches. The default size is 4096.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "int" },
- { "name": "camel.main.streamCachingEnabled", "description": "Sets whether stream caching is enabled or not. Default is false.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": "false" },
+ { "name": "camel.main.streamCachingEnabled", "description": "Sets whether stream caching is enabled or not. While stream types (like StreamSource, InputStream and Reader) are commonly used in messaging for performance reasons, they also have an important drawback: they can only be read once. In order to be able to work with message content multiple times, the stream needs to be cached. Streams are cached in memory. However, for large stream messages (over 128 KB by default) will be c [...]
{ "name": "camel.main.streamCachingRemoveSpoolDirectoryWhenStopping", "description": "Whether to remove stream caching temporary directory when stopping. This option is default true.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "boolean", "javaType": "boolean", "defaultValue": true },
{ "name": "camel.main.streamCachingSpoolCipher", "description": "Sets a stream caching cipher name to use when spooling to disk to write with encryption. By default the data is not encrypted.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
{ "name": "camel.main.streamCachingSpoolDirectory", "description": "Sets the stream caching spool (temporary) directory to use for overflow and spooling to disk. If no spool directory has been explicit configured, then a temporary directory is created in the java.io.tmpdir directory.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
diff --git a/core/camel-main/src/main/docs/main.adoc b/core/camel-main/src/main/docs/main.adoc
index c4045f9f28c..a9c5d2f5712 100644
--- a/core/camel-main/src/main/docs/main.adoc
+++ b/core/camel-main/src/main/docs/main.adoc
@@ -118,7 +118,7 @@ The camel.main supports 112 options, which are listed below.
| *camel.main.startupSummaryLevel* | Controls the level of information logged during startup (and shutdown) of CamelContext. | Default | StartupSummaryLevel
| *camel.main.streamCachingAny{zwsp}SpoolRules* | Sets whether if just any of the org.apache.camel.spi.StreamCachingStrategy.SpoolRule rules returns true then shouldSpoolCache(long) returns true, to allow spooling to disk. If this option is false, then all the org.apache.camel.spi.StreamCachingStrategy.SpoolRule must return true. The default value is false which means that all the rules must return true. | false | boolean
| *camel.main.streamCachingBuffer{zwsp}Size* | Sets the stream caching buffer size to use when allocating in-memory buffers used for in-memory stream caches. The default size is 4096. | | int
-| *camel.main.streamCaching{zwsp}Enabled* | Sets whether stream caching is enabled or not. Default is false. | false | boolean
+| *camel.main.streamCaching{zwsp}Enabled* | Sets whether stream caching is enabled or not. While stream types (like StreamSource, InputStream and Reader) are commonly used in messaging for performance reasons, they also have an important drawback: they can only be read once. In order to be able to work with message content multiple times, the stream needs to be cached. Streams are cached in memory. However, for large stream messages (over 128 KB by default) will be cached in a temporary [...]
| *camel.main.streamCachingRemove{zwsp}SpoolDirectoryWhenStopping* | Whether to remove stream caching temporary directory when stopping. This option is default true. | true | boolean
| *camel.main.streamCachingSpool{zwsp}Cipher* | Sets a stream caching cipher name to use when spooling to disk to write with encryption. By default the data is not encrypted. | | String
| *camel.main.streamCachingSpool{zwsp}Directory* | Sets the stream caching spool (temporary) directory to use for overflow and spooling to disk. If no spool directory has been explicit configured, then a temporary directory is created in the java.io.tmpdir directory. | | String
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
index 09fde0016bb..d1245f02606 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
@@ -56,7 +56,7 @@ public abstract class DefaultConfigurationProperties<T> {
private boolean devConsoleEnabled;
private boolean modeline;
private int logDebugMaxChars;
- private boolean streamCachingEnabled;
+ private boolean streamCachingEnabled = true;
private String streamCachingSpoolDirectory;
private String streamCachingSpoolCipher;
private long streamCachingSpoolThreshold;
@@ -426,7 +426,15 @@ public abstract class DefaultConfigurationProperties<T> {
/**
* Sets whether stream caching is enabled or not.
*
- * Default is false.
+ * While stream types (like StreamSource, InputStream and Reader) are commonly used in messaging for performance
+ * reasons, they also have an important drawback: they can only be read once. In order to be able to work with
+ * message content multiple times, the stream needs to be cached.
+ *
+ * Streams are cached in memory. However, for large stream messages (over 128 KB by default) will be cached in a
+ * temporary file instead, and Camel will handle deleting the temporary file once the cached stream is no longer
+ * necessary.
+ *
+ * Default is true.
*/
public void setStreamCachingEnabled(boolean streamCachingEnabled) {
this.streamCachingEnabled = streamCachingEnabled;
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedNonManagedServiceTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedNonManagedServiceTest.java
index 22e9d5931f4..20c33f244d1 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedNonManagedServiceTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedNonManagedServiceTest.java
@@ -34,7 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
@DisabledOnOs(OS.AIX)
public class ManagedNonManagedServiceTest extends ManagementTestSupport {
- private static final int SERVICES = 13;
+ private static final int SERVICES = 14;
@Test
public void testService() throws Exception {
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedProducerRouteAddRemoveRegisterAlwaysTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedProducerRouteAddRemoveRegisterAlwaysTest.java
index f18d3c6b4bc..57a0b4c33d5 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedProducerRouteAddRemoveRegisterAlwaysTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedProducerRouteAddRemoveRegisterAlwaysTest.java
@@ -36,7 +36,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@DisabledOnOs(OS.AIX)
public class ManagedProducerRouteAddRemoveRegisterAlwaysTest extends ManagementTestSupport {
- private static final int SERVICES = 13;
+ private static final int SERVICES = 14;
@Override
protected CamelContext createCamelContext() throws Exception {
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteAddRemoveTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteAddRemoveTest.java
index ac592b4053a..abeda66326b 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteAddRemoveTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteAddRemoveTest.java
@@ -40,7 +40,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@DisabledOnOs(OS.AIX)
public class ManagedRouteAddRemoveTest extends ManagementTestSupport {
- private static final int SERVICES = 13;
+ private static final int SERVICES = 14;
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedTypeConverterRegistryTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedTypeConverterRegistryTest.java
index 82291805425..7ada1440bd3 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedTypeConverterRegistryTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedTypeConverterRegistryTest.java
@@ -80,7 +80,7 @@ public class ManagedTypeConverterRegistryTest extends ManagementTestSupport {
failed = (Long) mbeanServer.getAttribute(name, "FailedCounter");
assertEquals(0, failed.intValue());
miss = (Long) mbeanServer.getAttribute(name, "MissCounter");
- assertEquals(0, miss.intValue());
+ assertEquals(2, miss.intValue()); // stream caching misses
// reset
mbeanServer.invoke(name, "resetTypeConversionCounters", null, null);
@@ -96,7 +96,7 @@ public class ManagedTypeConverterRegistryTest extends ManagementTestSupport {
failed = (Long) mbeanServer.getAttribute(name, "FailedCounter");
assertEquals(1, failed.intValue());
miss = (Long) mbeanServer.getAttribute(name, "MissCounter");
- assertEquals(0, miss.intValue());
+ assertEquals(1, miss.intValue()); // stream caching misses
// reset
mbeanServer.invoke(name, "resetTypeConversionCounters", null, null);
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/MessageHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/MessageHelper.java
index 5d3a132391d..a1d285a6784 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/MessageHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/MessageHelper.java
@@ -83,7 +83,8 @@ public final class MessageHelper {
}
// we need to favor using stream cache so the body can be re-read later
- StreamCache newBody = message.getBody(StreamCache.class);
+ StreamCache newBody = message.getExchange().getContext().getTypeConverter().tryConvertTo(StreamCache.class,
+ message.getExchange(), body);
if (newBody != null) {
message.setBody(newBody);
}
diff --git a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
index 97357a38e6f..981c308a9ca 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_17.adoc
@@ -6,6 +6,37 @@ from both 3.0 to 3.1 and 3.1 to 3.2.
== Upgrading Camel 3.16 to 3.17
+=== camel-core
+
+We have enabled xref:stream-caching.adoc[Stream Caching] by default on `CamelContext`. The reason is that Camel users
+may often hit this problem without knowing what is causing this, and blaming it on Apache Camel.
+
+This means that Camel would automatically cache message bodies that are based on `java.util.InputStream`
+such as often seen with HTTP components. This makes it safe to log the message body, and elsewhere.
+
+The cost is a tiny overhead in the routing engine as Camel will automatic type convert
+to `StreamCache` if needed. Users can turn this off:
+
+[source,java]
+----
+CamelContext context = ...
+context.setStreamCaching(false);
+----
+
+Or via Camel Main / Camel K / Quarkus:
+
+[source,properties]
+----
+camel.main.streamCachingEnabled=false
+----
+
+Or in Spring Boot
+
+[source,properties]
+----
+camel.springboot.streamCachingEnabled=false
+----
+
=== camel-health
Camel now reports DOWN when Camel is being stopped, during the graceful shutdown process.