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 9a833d26edba CAMEL-24321: Allow stream caching spool directory to be
resolved per-Exchange
9a833d26edba is described below
commit 9a833d26edba330e4fe727e3ebf75e6ec20d2c5f
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Aug 6 21:34:12 2026 +0200
CAMEL-24321: Allow stream caching spool directory to be resolved
per-Exchange
Add resolveSpoolDirectory(Exchange) default method to StreamCachingStrategy,
allowing custom implementations to direct each route's spooled bytes to a
separate directory. The default delegates to getSpoolDirectory() for full
backward compatibility.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/spi/StreamCachingStrategy.java | 16 ++
.../CachedOutputStreamPerExchangeSpoolDirTest.java | 274 +++++++++++++++++++++
.../camel/converter/stream/CachedOutputStream.java | 4 +-
.../converter/stream/FileInputStreamCache.java | 5 +-
.../modules/ROOT/pages/stream-caching.adoc | 23 ++
5 files changed, 319 insertions(+), 3 deletions(-)
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/StreamCachingStrategy.java
b/core/camel-api/src/main/java/org/apache/camel/spi/StreamCachingStrategy.java
index 631d844425c4..90f1ae7ed179 100644
---
a/core/camel-api/src/main/java/org/apache/camel/spi/StreamCachingStrategy.java
+++
b/core/camel-api/src/main/java/org/apache/camel/spi/StreamCachingStrategy.java
@@ -196,6 +196,22 @@ public interface StreamCachingStrategy extends
StaticService {
@Nullable
File getSpoolDirectory();
+ /**
+ * Resolves the spool directory to use for the given {@link Exchange}.
This allows custom implementations to direct
+ * each route's spooled bytes to a separate directory, for example to
isolate spool data per route or to monitor
+ * spool usage at the route level.
+ * <p/>
+ * The default implementation delegates to {@link #getSpoolDirectory()},
which returns the single context-wide spool
+ * directory. Custom implementations may override this to return a
per-route or per-exchange directory.
+ *
+ * @param exchange the exchange being processed
+ * @return the spool directory to use for this exchange, or
{@code null} if spooling is not configured
+ * @since 4.22
+ */
+ default @Nullable File resolveSpoolDirectory(Exchange exchange) {
+ return getSpoolDirectory();
+ }
+
void setSpoolDirectory(String path);
/**
diff --git
a/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamPerExchangeSpoolDirTest.java
b/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamPerExchangeSpoolDirTest.java
new file mode 100644
index 000000000000..23ca409c40c5
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/converter/stream/CachedOutputStreamPerExchangeSpoolDirTest.java
@@ -0,0 +1,274 @@
+/*
+ * 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.converter.stream;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.StringJoiner;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.StreamCache;
+import org.apache.camel.impl.engine.DefaultStreamCachingStrategy;
+import org.apache.camel.impl.engine.DefaultUnitOfWork;
+import org.apache.camel.spi.UnitOfWork;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.util.IOHelper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for per-Exchange spool directory resolution via
+ * {@link
org.apache.camel.spi.StreamCachingStrategy#resolveSpoolDirectory(Exchange)}.
+ */
+class CachedOutputStreamPerExchangeSpoolDirTest extends ContextTestSupport {
+
+ private static final String TEST_STRING = "This is a test string that is
long enough to exceed the spool threshold"
+ + "
aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa ";
+
+ private Exchange exchange;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.setStreamCaching(true);
+
context.getStreamCachingStrategy().setSpoolDirectory(testDirectory().toFile());
+ context.getStreamCachingStrategy().setSpoolEnabled(true);
+ context.getStreamCachingStrategy().setSpoolThreshold(16);
+ return context;
+ }
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+
+ exchange = new DefaultExchange(context);
+ UnitOfWork uow = new DefaultUnitOfWork(exchange);
+ exchange.getExchangeExtension().setUnitOfWork(uow);
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ private static String toString(InputStream input) throws IOException {
+ BufferedReader reader = IOHelper.buffered(new
InputStreamReader(input));
+ StringJoiner builder = new StringJoiner(", ");
+ while (true) {
+ String line = reader.readLine();
+ if (line == null) {
+ return builder.toString();
+ }
+ builder.add(line);
+ }
+ }
+
+ @Test
+ void testDefaultResolveSpoolDirectoryReturnsSameDirectory() throws
Exception {
+ context.start();
+
+ // The default resolveSpoolDirectory should return the same directory
as getSpoolDirectory
+ File spoolDir = context.getStreamCachingStrategy().getSpoolDirectory();
+ File resolvedDir =
context.getStreamCachingStrategy().resolveSpoolDirectory(exchange);
+ assertThat(resolvedDir).isEqualTo(spoolDir);
+
+ // Verify spooling still works with default behavior
+ CachedOutputStream cos = new CachedOutputStream(exchange);
+ cos.write(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+
+ StreamCache cache = cos.newStreamCache();
+ assertThat(cache).isInstanceOf(FileInputStreamCache.class);
+ String content = toString((InputStream) cache);
+ assertThat(content).isEqualTo(TEST_STRING);
+
+ ((InputStream) cache).close();
+ exchange.getUnitOfWork().done(exchange);
+ IOHelper.close(cos);
+ }
+
+ @Test
+ void testPerExchangeSpoolDirectoryResolution() throws Exception {
+ // Use a custom strategy that returns a per-route subdirectory
+ File baseDir = testDirectory().toFile();
+ DefaultStreamCachingStrategy customStrategy = new
DefaultStreamCachingStrategy() {
+ @Override
+ public File resolveSpoolDirectory(Exchange exchange) {
+ String routeId = exchange.getFromRouteId();
+ if (routeId != null) {
+ return new File(getSpoolDirectory(), routeId);
+ }
+ return getSpoolDirectory();
+ }
+ };
+ customStrategy.setCamelContext(context);
+ customStrategy.setEnabled(true);
+ customStrategy.setSpoolEnabled(true);
+ customStrategy.setSpoolDirectory(baseDir);
+ customStrategy.setSpoolThreshold(16);
+ context.setStreamCachingStrategy(customStrategy);
+
+ context.start();
+
+ // Create an exchange that simulates coming from a route
+ Exchange exchangeWithRoute = new DefaultExchange(context);
+ exchangeWithRoute.getExchangeExtension().setFromRouteId("routeA");
+ UnitOfWork uow = new DefaultUnitOfWork(exchangeWithRoute);
+ exchangeWithRoute.getExchangeExtension().setUnitOfWork(uow);
+
+ CachedOutputStream cos = new CachedOutputStream(exchangeWithRoute);
+ cos.write(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+
+ // The temp file should be in the per-route subdirectory
+ StreamCache cache = cos.newStreamCache();
+ assertThat(cache).isInstanceOf(FileInputStreamCache.class);
+
+ // Verify the subdirectory was created
+ File routeDir = new File(baseDir, "routeA");
+ assertThat(routeDir).exists().isDirectory();
+
+ // Verify files are in the per-route subdirectory
+ String[] files = routeDir.list();
+ assertThat(files).isNotNull().hasSize(1);
+ assertThat(files[0]).startsWith("cos");
+
+ // Verify content is correct
+ String content = toString((InputStream) cache);
+ assertThat(content).isEqualTo(TEST_STRING);
+
+ ((InputStream) cache).close();
+ exchangeWithRoute.getUnitOfWork().done(exchangeWithRoute);
+ IOHelper.close(cos);
+ }
+
+ @Test
+ void testPerExchangeSpoolDirWithDifferentRoutes() throws Exception {
+ // Use a custom strategy that returns a per-route subdirectory
+ File baseDir = testDirectory().toFile();
+ DefaultStreamCachingStrategy customStrategy = new
DefaultStreamCachingStrategy() {
+ @Override
+ public File resolveSpoolDirectory(Exchange exchange) {
+ String routeId = exchange.getFromRouteId();
+ if (routeId != null) {
+ return new File(getSpoolDirectory(), routeId);
+ }
+ return getSpoolDirectory();
+ }
+ };
+ customStrategy.setCamelContext(context);
+ customStrategy.setEnabled(true);
+ customStrategy.setSpoolEnabled(true);
+ customStrategy.setSpoolDirectory(baseDir);
+ customStrategy.setSpoolThreshold(16);
+ context.setStreamCachingStrategy(customStrategy);
+
+ context.start();
+
+ // Create exchange for route A
+ Exchange exchangeA = new DefaultExchange(context);
+ exchangeA.getExchangeExtension().setFromRouteId("routeA");
+ UnitOfWork uowA = new DefaultUnitOfWork(exchangeA);
+ exchangeA.getExchangeExtension().setUnitOfWork(uowA);
+
+ CachedOutputStream cosA = new CachedOutputStream(exchangeA);
+ cosA.write(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+
+ // Create exchange for route B
+ Exchange exchangeB = new DefaultExchange(context);
+ exchangeB.getExchangeExtension().setFromRouteId("routeB");
+ UnitOfWork uowB = new DefaultUnitOfWork(exchangeB);
+ exchangeB.getExchangeExtension().setUnitOfWork(uowB);
+
+ CachedOutputStream cosB = new CachedOutputStream(exchangeB);
+ cosB.write(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+
+ // Verify both subdirectories were created and contain spool files
+ File routeADir = new File(baseDir, "routeA");
+ File routeBDir = new File(baseDir, "routeB");
+
+ assertThat(routeADir).exists().isDirectory();
+ assertThat(routeBDir).exists().isDirectory();
+
+ String[] filesA = routeADir.list();
+ String[] filesB = routeBDir.list();
+ assertThat(filesA).isNotNull().hasSize(1);
+ assertThat(filesB).isNotNull().hasSize(1);
+
+ // Verify content of both streams
+ StreamCache cacheA = cosA.newStreamCache();
+ StreamCache cacheB = cosB.newStreamCache();
+ assertThat(toString((InputStream) cacheA)).isEqualTo(TEST_STRING);
+ assertThat(toString((InputStream) cacheB)).isEqualTo(TEST_STRING);
+
+ ((InputStream) cacheA).close();
+ ((InputStream) cacheB).close();
+ exchangeA.getUnitOfWork().done(exchangeA);
+ exchangeB.getUnitOfWork().done(exchangeB);
+ IOHelper.close(cosA);
+ IOHelper.close(cosB);
+ }
+
+ @Test
+ void testPerExchangeSpoolDirFallsBackWhenNoRouteId() throws Exception {
+ // Use a custom strategy that returns a per-route subdirectory
+ File baseDir = testDirectory().toFile();
+ DefaultStreamCachingStrategy customStrategy = new
DefaultStreamCachingStrategy() {
+ @Override
+ public File resolveSpoolDirectory(Exchange exchange) {
+ String routeId = exchange.getFromRouteId();
+ if (routeId != null) {
+ return new File(getSpoolDirectory(), routeId);
+ }
+ return getSpoolDirectory();
+ }
+ };
+ customStrategy.setCamelContext(context);
+ customStrategy.setEnabled(true);
+ customStrategy.setSpoolEnabled(true);
+ customStrategy.setSpoolDirectory(baseDir);
+ customStrategy.setSpoolThreshold(16);
+ context.setStreamCachingStrategy(customStrategy);
+
+ context.start();
+
+ // Exchange without fromRouteId should fall back to base directory
+ CachedOutputStream cos = new CachedOutputStream(exchange);
+ cos.write(TEST_STRING.getBytes(StandardCharsets.UTF_8));
+
+ StreamCache cache = cos.newStreamCache();
+ assertThat(cache).isInstanceOf(FileInputStreamCache.class);
+
+ // Files should be in the base directory (no subdirectory)
+ String[] files = baseDir.list();
+ assertThat(files).isNotNull().hasSizeGreaterThanOrEqualTo(1);
+
+ String content = toString((InputStream) cache);
+ assertThat(content).isEqualTo(TEST_STRING);
+
+ ((InputStream) cache).close();
+ exchange.getUnitOfWork().done(exchange);
+ IOHelper.close(cos);
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
b/core/camel-support/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
index 3a160d49c60a..4863a2fd102f 100644
---
a/core/camel-support/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
+++
b/core/camel-support/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
@@ -43,6 +43,7 @@ import org.apache.camel.util.IOHelper;
public class CachedOutputStream extends OutputStream {
private final StreamCachingStrategy strategy;
+ private final Exchange exchange;
private OutputStream currentStream;
private boolean inMemory = true;
private int totalLength;
@@ -55,6 +56,7 @@ public class CachedOutputStream extends OutputStream {
public CachedOutputStream(Exchange exchange, final boolean
closedOnCompletion) {
this.closedOnCompletion = closedOnCompletion;
+ this.exchange = exchange;
this.tempFileManager = new TempFileManager(closedOnCompletion);
this.tempFileManager.addExchange(exchange);
this.strategy = exchange.getContext().getStreamCachingStrategy();
@@ -159,7 +161,7 @@ public class CachedOutputStream extends OutputStream {
CachedByteArrayOutputStream bout = (CachedByteArrayOutputStream)
currentStream;
try {
// creates a tmp file and a file output stream
- currentStream = tempFileManager.createOutputStream(strategy);
+ currentStream = tempFileManager.createOutputStream(strategy,
exchange);
IOHelper.copy(bout.newInputStreamCache(), currentStream,
strategy.getBufferSize());
} finally {
// ensure flag is flipped to file based
diff --git
a/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
b/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
index 3992b02c5224..d1d16a403aa7 100644
---
a/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
+++
b/core/camel-support/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
@@ -303,7 +303,7 @@ public final class FileInputStreamCache extends InputStream
implements StreamCac
}
}
- OutputStream createOutputStream(StreamCachingStrategy strategy) throws
IOException {
+ OutputStream createOutputStream(StreamCachingStrategy strategy,
Exchange exchange) throws IOException {
// should only be called once
if (tempFile != null) {
throw new IllegalStateException("The method
'createOutputStream' can only be called once!");
@@ -320,7 +320,8 @@ public final class FileInputStreamCache extends InputStream
implements StreamCac
LOG.error(error);
throw new IOException(error);
}
- tempFile = FileUtil.createTempFile("cos", ".tmp",
strategy.getSpoolDirectory());
+ File spoolDir = strategy.resolveSpoolDirectory(exchange);
+ tempFile = FileUtil.createTempFile("cos", ".tmp", spoolDir);
LOG.trace("Creating temporary stream cache file: {}", tempFile);
OutputStream out = new BufferedOutputStream(
diff --git a/docs/user-manual/modules/ROOT/pages/stream-caching.adoc
b/docs/user-manual/modules/ROOT/pages/stream-caching.adoc
index 00d7fc9f9be8..55f08813aa76 100644
--- a/docs/user-manual/modules/ROOT/pages/stream-caching.adoc
+++ b/docs/user-manual/modules/ROOT/pages/stream-caching.adoc
@@ -344,6 +344,29 @@
camel.main.streamCachingSpoolRules=#class:com.foo.MySpoolRule
====
+== Using custom spool directory per Exchange (advanced)
+
+By default, all spooled streams are written to a single spool directory. If
you want to control where each exchange's spool files are written (for example,
to isolate spool data per route), you can implement a custom
`StreamCachingStrategy` that overrides the `resolveSpoolDirectory` method.
+
+The `resolveSpoolDirectory` method is called for each exchange when a stream
is about to be spooled to disk. The default implementation returns the
configured `spoolDirectory`. Custom implementations can return a different
directory based on the exchange, for example using the route id as a
subdirectory:
+
+[source,java]
+----
+DefaultStreamCachingStrategy strategy = new DefaultStreamCachingStrategy() {
+ @Override
+ public File resolveSpoolDirectory(Exchange exchange) {
+ String routeId = exchange.getFromRouteId();
+ if (routeId != null) {
+ return new File(getSpoolDirectory(), routeId);
+ }
+ return getSpoolDirectory();
+ }
+};
+context.setStreamCachingStrategy(strategy);
+----
+
+The resolved directory is created automatically if it does not exist.
+
== Using StreamCachingProcessor
Since *Camel 4.11* this processor can be used to convert the current message
body to a `StreamCache`. This allows the body to be re-read multiple times and
can be placed at any point in a Camel route.
