camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/6] camel git commit: CAMEL-8688: Stream cache now keeps track on number of copies that was spooled to disk, so when the temp file is deleted its only deleted when no longer in use. A file can be shared if using wire tap etc. Thanks to Franz Forsthofer
Date Mon, 04 May 2015 17:59:41 GMT
CAMEL-8688: Stream cache now keeps track on number of copies that was spooled to disk, so when
the temp file is deleted its only deleted when no longer in use. A file can be shared if using
wire tap etc. Thanks to Franz Forsthofer for the patch.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a99f6d57
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a99f6d57
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a99f6d57

Branch: refs/heads/master
Commit: a99f6d5710c3068fa3fe841d4f80dc82deb1142b
Parents: cfdf185
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon May 4 19:09:51 2015 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon May 4 19:31:30 2015 +0200

----------------------------------------------------------------------
 .../main/java/org/apache/camel/StreamCache.java |   6 +-
 .../stream/ByteArrayInputStreamCache.java       |   3 +-
 .../converter/stream/CachedOutputStream.java    | 121 ++-----------
 .../converter/stream/FileInputStreamCache.java  | 173 +++++++++++++++++--
 .../converter/stream/InputStreamCache.java      |   3 +-
 .../camel/converter/stream/ReaderCache.java     |   3 +-
 .../camel/converter/stream/SourceCache.java     |   3 +-
 .../converter/stream/StreamSourceCache.java     |   6 +-
 .../camel/processor/MulticastProcessor.java     |   2 +-
 .../camel/processor/WireTapProcessor.java       |   2 +-
 .../processor/WireTapStreamCachingTest.java     |  19 +-
 .../apache/camel/util/MessageHelperTest.java    |   2 +-
 .../apache/camel/processor/twoCharacters.txt    |   1 +
 .../http/NettyChannelBufferStreamCache.java     |   3 +-
 14 files changed, 207 insertions(+), 140 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/StreamCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/StreamCache.java b/camel-core/src/main/java/org/apache/camel/StreamCache.java
index ecd9736..29f4284 100644
--- a/camel-core/src/main/java/org/apache/camel/StreamCache.java
+++ b/camel-core/src/main/java/org/apache/camel/StreamCache.java
@@ -29,7 +29,7 @@ import java.io.OutputStream;
  * The Camel routing engine uses the {@link org.apache.camel.processor.CamelInternalProcessor.StreamCachingAdvice}
  * to apply the stream cache during routing.
  * <p/>
- * It is recommended in the {@link #copy()} method to let the copied stream start from the
start. If the implementation
+ * It is recommended in the {@link #copy(Exchange)} method to let the copied stream start
from the start. If the implementation
  * does not support copy, then return <tt>null</tt>.
  *
  * @version 
@@ -60,10 +60,12 @@ public interface StreamCache {
      * Implementations note: A copy of the stream is recommended to read from the start
      * of the stream.
      *
+     * @param exchange exchange in which the stream cache object is used; 
+     *                 can be used to delete resources of the stream cache when the exchange
is completed
      * @return a copy, or <tt>null</tt> if copy is not possible
      * @throws java.io.IOException is thrown if the copy fails
      */
-    StreamCache copy() throws IOException;
+    StreamCache copy(Exchange exchange) throws IOException;
 
     /**
      * Whether this {@link StreamCache} is in memory only or

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/ByteArrayInputStreamCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/ByteArrayInputStreamCache.java
b/camel-core/src/main/java/org/apache/camel/converter/stream/ByteArrayInputStreamCache.java
index 3b1dacf..9375ee3 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/ByteArrayInputStreamCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/ByteArrayInputStreamCache.java
@@ -22,6 +22,7 @@ import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
 import org.apache.camel.util.IOHelper;
 
@@ -51,7 +52,7 @@ public class ByteArrayInputStreamCache extends FilterInputStream implements
Stre
         IOHelper.copyAndCloseInput(in, os);
     }
 
-    public StreamCache copy() throws IOException {
+    public StreamCache copy(Exchange exchange) throws IOException {
         if (byteArrayForCopy == null) {
             ByteArrayOutputStream baos = new ByteArrayOutputStream(in.available());
             IOHelper.copy(in, baos);

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
b/camel-core/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
index 639e339..d722baf 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/CachedOutputStream.java
@@ -16,29 +16,15 @@
  */
 package org.apache.camel.converter.stream;
 
-import java.io.BufferedOutputStream;
 import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.security.GeneralSecurityException;
-
-import javax.crypto.CipherOutputStream;
 
 import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
-import org.apache.camel.converter.stream.FileInputStreamCache.FileInputStreamCloser;
+import org.apache.camel.converter.stream.FileInputStreamCache.TempFileManager;
 import org.apache.camel.spi.StreamCachingStrategy;
-import org.apache.camel.spi.Synchronization;
-import org.apache.camel.spi.UnitOfWork;
-import org.apache.camel.support.SynchronizationAdapter;
-import org.apache.camel.util.FileUtil;
-import org.apache.camel.util.ObjectHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * This output stream will store the content into a File if the stream context size is exceed
the
@@ -50,7 +36,7 @@ import org.slf4j.LoggerFactory;
  * <p/>
  * You can get a cached input stream of this stream. The temp file which is created with
this 
  * output stream will be deleted when you close this output stream or the cached 
- * fileInputStream(s) is/are closed after the exchange is completed.
+ * fileInputStream(s) is/are closed after all the exchanges using the temp file are completed.
  */
 public class CachedOutputStream extends OutputStream {
     @Deprecated
@@ -61,16 +47,12 @@ public class CachedOutputStream extends OutputStream {
     public static final String TEMP_DIR = "CamelCachedOutputStreamOutputDirectory";
     @Deprecated
     public static final String CIPHER_TRANSFORMATION = "CamelCachedOutputStreamCipherTransformation";
-    private static final Logger LOG = LoggerFactory.getLogger(CachedOutputStream.class);
 
     private final StreamCachingStrategy strategy;
     private OutputStream currentStream;
     private boolean inMemory = true;
     private int totalLength;
-    private File tempFile;
-    private FileInputStreamCache fileInputStreamCache;
-    private final FileInputStreamCloser fileInputStreamCloser = new FileInputStreamCloser();
-    private CipherPair ciphers;
+    private final TempFileManager tempFileManager;
     private final boolean closedOnCompletion;
 
     public CachedOutputStream(Exchange exchange) {
@@ -79,44 +61,10 @@ public class CachedOutputStream extends OutputStream {
 
     public CachedOutputStream(Exchange exchange, final boolean closedOnCompletion) {
         this.closedOnCompletion = closedOnCompletion;
+        tempFileManager = new TempFileManager(closedOnCompletion);
+        tempFileManager.addExchange(exchange);
         this.strategy = exchange.getContext().getStreamCachingStrategy();
         currentStream = new CachedByteArrayOutputStream(strategy.getBufferSize());
-        if (closedOnCompletion) {
-            // add on completion so we can cleanup after the exchange is done such as deleting
temporary files
-            Synchronization onCompletion = new SynchronizationAdapter() {
-                @Override
-                public void onDone(Exchange exchange) {
-                    try {
-                        closeFileInputStreams();
-                        close();
-                        try {
-                            cleanUpTempFile();
-                        } catch (Exception e) {
-                            LOG.warn("Error deleting temporary cache file: " + tempFile +
". This exception will be ignored.", e);
-                        }
-                    } catch (Exception e) {
-                        LOG.warn("Error closing streams. This exception will be ignored.",
e);
-                    }
-                }
-
-                @Override
-                public String toString() {
-                    return "OnCompletion[CachedOutputStream]";
-                }
-            };
-
-            UnitOfWork streamCacheUnitOfWork = exchange.getProperty(Exchange.STREAM_CACHE_UNIT_OF_WORK,
UnitOfWork.class);
-            if (streamCacheUnitOfWork != null) {
-                // The stream cache must sometimes not be closed when the exchange is deleted.
This is for example the
-                // case in the splitter and multi-cast case with AggregationStrategy where
the result of the sub-routes
-                // are aggregated later in the main route. Here, the cached streams of the
sub-routes must be closed with
-                // the Unit of Work of the main route.
-                streamCacheUnitOfWork.addSynchronization(onCompletion);
-            } else {
-                // add on completion so we can cleanup after the exchange is done such as
deleting temporary files
-                exchange.addOnCompletion(onCompletion);
-            }
-        }
     }
 
     public void flush() throws IOException {
@@ -127,12 +75,8 @@ public class CachedOutputStream extends OutputStream {
         currentStream.close();
         // need to clean up the temp file this time
         if (!closedOnCompletion) {
-            closeFileInputStreams();
-            try {
-                cleanUpTempFile();
-            } catch (Exception e) {
-                LOG.warn("Error deleting temporary cache file: " + tempFile + ". This exception
will be ignored.", e);
-            }
+            tempFileManager.closeFileInputStreams();
+            tempFileManager.cleanUpTempFile();
         }
     }
 
@@ -206,40 +150,17 @@ public class CachedOutputStream extends OutputStream {
                 throw new IllegalStateException("CurrentStream should be an instance of CachedByteArrayOutputStream
but is: " + currentStream.getClass().getName());
             }
         } else {
-            try {
-                if (fileInputStreamCache == null) {
-                    fileInputStreamCache = new FileInputStreamCache(tempFile, ciphers, fileInputStreamCloser);
-                }
-                return fileInputStreamCache;
-            } catch (FileNotFoundException e) {
-                throw new IOException("Cached file " + tempFile + " not found", e);
-            }
+            return tempFileManager.newStreamCache();
         }
     }
     
-    private void closeFileInputStreams() {
-        fileInputStreamCloser.close();
-        fileInputStreamCache = null;
-    } 
-
-    private void cleanUpTempFile() {
-        // cleanup temporary file
-        if (tempFile != null) {
-            FileUtil.deleteFile(tempFile);
-            tempFile = null;
-        }
-    }
 
     private void pageToFileStream() throws IOException {
         flush();
-
         ByteArrayOutputStream bout = (ByteArrayOutputStream)currentStream;
-        tempFile = FileUtil.createTempFile("cos", ".tmp", strategy.getSpoolDirectory());
-
-        LOG.trace("Creating temporary stream cache file: {}", tempFile);
-
         try {
-            currentStream = createOutputStream(tempFile);
+            // creates an tmp file and a file output stream
+            currentStream = tempFileManager.createOutputStream(strategy);
             bout.writeTo(currentStream);
         } finally {
             // ensure flag is flipped to file based
@@ -291,26 +212,4 @@ public class CachedOutputStream extends OutputStream {
         }
     }
 
-    private OutputStream createOutputStream(File file) throws IOException {
-        OutputStream out = new BufferedOutputStream(new FileOutputStream(file));
-        if (ObjectHelper.isNotEmpty(strategy.getSpoolChiper())) {
-            try {
-                if (ciphers == null) {
-                    ciphers = new CipherPair(strategy.getSpoolChiper());
-                }
-            } catch (GeneralSecurityException e) {
-                throw new IOException(e.getMessage(), e);
-            }
-            out = new CipherOutputStream(out, ciphers.getEncryptor()) {
-                boolean closed;
-                public void close() throws IOException {
-                    if (!closed) {
-                        super.close();
-                        closed = true;
-                    }
-                }
-            };
-        }
-        return out;
-    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
b/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
index c6e99ac..a0d6501 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
@@ -17,44 +17,60 @@
 package org.apache.camel.converter.stream;
 
 import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.WritableByteChannel;
+import java.security.GeneralSecurityException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import javax.crypto.CipherInputStream;
+import javax.crypto.CipherOutputStream;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.RuntimeCamelException;
 import org.apache.camel.StreamCache;
+import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.UnitOfWork;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.util.FileUtil;
 import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link StreamCache} for {@link File}s
  */
 public final class FileInputStreamCache extends InputStream implements StreamCache {
     private InputStream stream;
+    private final long length;
+    private final FileInputStreamCache.TempFileManager tempFileManager;
     private final File file;
     private final CipherPair ciphers;
-    private final long length;
-    private final FileInputStreamCache.FileInputStreamCloser closer;
 
+    /** Only for testing purposes.*/
     public FileInputStreamCache(File file) throws FileNotFoundException {
-        this(file, null, new FileInputStreamCloser());
+        this(new TempFileManager(file, true));
     }
     
-    FileInputStreamCache(File file, CipherPair ciphers, FileInputStreamCloser closer) throws
FileNotFoundException {
-        this.file = file;
+    FileInputStreamCache(TempFileManager closer) throws FileNotFoundException {
+        this.file = closer.getTempFile();
         this.stream = null;
-        this.ciphers = ciphers;
+        this.ciphers = closer.getCiphers();
         this.length = file.length();
-        this.closer = closer;
-        this.closer.add(this);
+        this.tempFileManager = closer;
+        this.tempFileManager.add(this);
     }
     
     @Override
@@ -99,8 +115,9 @@ public final class FileInputStreamCache extends InputStream implements
StreamCac
         }
     }
 
-    public StreamCache copy() throws IOException {
-        FileInputStreamCache copy = new FileInputStreamCache(file, ciphers, closer);
+    public StreamCache copy(Exchange exchange) throws IOException {
+        tempFileManager.addExchange(exchange);
+        FileInputStreamCache copy = new FileInputStreamCache(tempFileManager);
         return copy;
     }
 
@@ -146,16 +163,37 @@ public final class FileInputStreamCache extends InputStream implements
StreamCac
     }
 
     /** 
-     * Collects all FileInputStreamCache instances of a temporary file which must be closed
-     * at the end of the route.
+     * Manages the temporary file for the file input stream caches.
+     * 
+     * Collects all FileInputStreamCache instances of the temporary file.
+     * Counts the number of exchanges which have a FileInputStreamCache  instance of the
temporary file.
+     * Deletes the temporary file, if all exchanges are done.
      * 
      * @see CachedOutputStream
      */
-    static class FileInputStreamCloser {
+    static class TempFileManager {
+        
+        private static final Logger LOG = LoggerFactory.getLogger(TempFileManager.class);
+        /** Indicator whether the file input stream caches are closed on completion of the
exchanges. */
+        private final boolean closedOnCompletion;
+        private AtomicInteger exchangeCounter = new AtomicInteger();
+        private File tempFile;
+        private OutputStream outputStream; // file output stream
+        private CipherPair ciphers;
         
-        // there can be several input streams, for example in the multi-cast parallel processing
+        // there can be several input streams, for example in the multi-cast, or wiretap
parallel processing
         private List<FileInputStreamCache> fileInputStreamCaches;
+
+        /** Only for testing.*/
+        private TempFileManager(File file, boolean closedOnCompletion) {
+            this(closedOnCompletion);
+            this.tempFile = file;
+        }
         
+        TempFileManager(boolean closedOnCompletion) {
+            this.closedOnCompletion = closedOnCompletion;
+        }
+                
         /** Adds a FileInputStreamCache instance to the closer.
          * <p>
          * Must be synchronized, because can be accessed by several threads. 
@@ -167,14 +205,119 @@ public final class FileInputStreamCache extends InputStream implements
StreamCac
             fileInputStreamCaches.add(fileInputStreamCache);
         }
         
-        void close() {
+        void addExchange(Exchange exchange) {
+            if (closedOnCompletion) {
+                exchangeCounter.incrementAndGet();
+                // add on completion so we can cleanup after the exchange is done such as
deleting temporary files
+                Synchronization onCompletion = new SynchronizationAdapter() {
+                    @Override
+                    public void onDone(Exchange exchange) {
+                        int actualExchanges = exchangeCounter.decrementAndGet();
+                        if (actualExchanges == 0) {
+                            // only one exchange (one thread) left, therefore we must not
synchronize the following lines of code
+                            try {                              
+                                closeFileInputStreams();
+                                if (outputStream != null) {
+                                    outputStream.close();
+                                }
+                                try {
+                                    cleanUpTempFile();
+                                } catch (Exception e) {
+                                    LOG.warn("Error deleting temporary cache file: " + tempFile
+ ". This exception will be ignored.", e);
+                                }
+                            } catch (Exception e) {
+                                LOG.warn("Error closing streams. This exception will be ignored.",
e);
+                            }
+                        }
+                    }
+
+                    @Override
+                    public String toString() {
+                        return "OnCompletion[CachedOutputStream]";
+                    }
+                };
+                UnitOfWork streamCacheUnitOfWork = exchange.getProperty(Exchange.STREAM_CACHE_UNIT_OF_WORK,
UnitOfWork.class);
+                if (streamCacheUnitOfWork != null) {
+                    // The stream cache must sometimes not be closed when the exchange is
deleted. This is for example the
+                    // case in the splitter and multi-cast case with AggregationStrategy
where the result of the sub-routes
+                    // are aggregated later in the main route. Here, the cached streams of
the sub-routes must be closed with
+                    // the Unit of Work of the main route.
+                    streamCacheUnitOfWork.addSynchronization(onCompletion);
+                } else {
+                    // add on completion so we can cleanup after the exchange is done such
as deleting temporary files
+                    exchange.addOnCompletion(onCompletion);
+                }
+            }
+        }
+        
+        OutputStream createOutputStream(StreamCachingStrategy strategy) throws IOException
{
+            // should only be called once
+            if (tempFile != null) {
+                throw new IllegalStateException("The method 'createOutputStream' can only
be called once!");
+            }
+            tempFile = FileUtil.createTempFile("cos", ".tmp", strategy.getSpoolDirectory());
+
+            LOG.trace("Creating temporary stream cache file: {}", tempFile);
+            OutputStream out = new BufferedOutputStream(new FileOutputStream(tempFile));
+            if (ObjectHelper.isNotEmpty(strategy.getSpoolChiper())) {
+                try {
+                    if (ciphers == null) {
+                        ciphers = new CipherPair(strategy.getSpoolChiper());
+                    }
+                } catch (GeneralSecurityException e) {
+                    throw new IOException(e.getMessage(), e);
+                }
+                out = new CipherOutputStream(out, ciphers.getEncryptor()) {
+                    boolean closed;
+                    public void close() throws IOException {
+                        if (!closed) {
+                            super.close();
+                            closed = true;
+                        }
+                    }
+                };
+            }
+            outputStream = out;
+            return out;
+        }
+        
+        FileInputStreamCache newStreamCache() throws IOException {
+            try {
+                return new FileInputStreamCache(this);
+            } catch (FileNotFoundException e) {
+                throw new IOException("Cached file " + tempFile + " not found", e);
+            }
+        }
+        
+        void closeFileInputStreams() {
             if (fileInputStreamCaches != null) {
                 for (FileInputStreamCache fileInputStreamCache : fileInputStreamCaches) {
                     fileInputStreamCache.close();
                 }
                 fileInputStreamCaches.clear();
             }
+        } 
+
+        void cleanUpTempFile() {
+            // cleanup temporary file
+            try {
+                if (tempFile != null) {
+                    FileUtil.deleteFile(tempFile);
+                    tempFile = null;
+                }
+            } catch (Exception e) {
+                LOG.warn("Error deleting temporary cache file: " + tempFile + ". This exception
will be ignored.", e);
+            }
+        }
+        
+        File getTempFile() {
+            return tempFile;
+        }
+        
+        CipherPair getCiphers() {
+            return ciphers;
         }
+        
     }
 
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/InputStreamCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/InputStreamCache.java
b/camel-core/src/main/java/org/apache/camel/converter/stream/InputStreamCache.java
index ba7f12e..78422a7 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/InputStreamCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/InputStreamCache.java
@@ -20,6 +20,7 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
 
 /**
@@ -40,7 +41,7 @@ public final class InputStreamCache extends ByteArrayInputStream implements
Stre
         os.write(buf, pos, count - pos);
     }
 
-    public StreamCache copy() {
+    public StreamCache copy(Exchange exchange) {
         return new InputStreamCache(buf, count);
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/ReaderCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/ReaderCache.java b/camel-core/src/main/java/org/apache/camel/converter/stream/ReaderCache.java
index bed761c..2890945 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/ReaderCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/ReaderCache.java
@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.io.OutputStream;
 import java.io.StringReader;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
 
 /**
@@ -51,7 +52,7 @@ public class ReaderCache extends StringReader implements StreamCache {
         os.write(data.getBytes());
     }
 
-    public StreamCache copy() throws IOException {
+    public StreamCache copy(Exchange exchange) throws IOException {
         return new ReaderCache(data);
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/SourceCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/SourceCache.java b/camel-core/src/main/java/org/apache/camel/converter/stream/SourceCache.java
index 16f8422..4f00eb4 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/SourceCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/SourceCache.java
@@ -19,6 +19,7 @@ package org.apache.camel.converter.stream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
 import org.apache.camel.StringSource;
 import org.apache.camel.util.IOHelper;
@@ -44,7 +45,7 @@ public final class SourceCache extends StringSource implements StreamCache
{
         IOHelper.copy(getInputStream(), os);
     }
 
-    public StreamCache copy() throws IOException {
+    public StreamCache copy(Exchange exchange) throws IOException {
         return new SourceCache(getText());
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/converter/stream/StreamSourceCache.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/StreamSourceCache.java
b/camel-core/src/main/java/org/apache/camel/converter/stream/StreamSourceCache.java
index 499f799..a7edfc9 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/StreamSourceCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/StreamSourceCache.java
@@ -83,12 +83,12 @@ public final class StreamSourceCache extends StreamSource implements StreamCache
         }
     }
 
-    public StreamCache copy() throws IOException {
+    public StreamCache copy(Exchange exchange) throws IOException {
         if (streamCache != null) {
-            return new StreamSourceCache(streamCache.copy());
+            return new StreamSourceCache(streamCache.copy(exchange));
         }
         if (readCache != null) {
-            return new StreamSourceCache(readCache.copy());
+            return new StreamSourceCache(readCache.copy(exchange));
         }
         return null;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index 334ceb1..e4a2ef8 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -953,7 +953,7 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
                 if (index > 0) {
                     // copy it otherwise parallel processing is not possible,
                     // because streams can only be read once
-                    StreamCache copiedStreamCache = streamCache.copy();
+                    StreamCache copiedStreamCache = streamCache.copy(copy);
                     if (copiedStreamCache != null) {
                         copy.getIn().setBody(copiedStreamCache);  
                     }

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
index a74e663..1d6b835 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
@@ -180,7 +180,7 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
         if (msg.getBody() instanceof StreamCache) {
             // in parallel processing case, the stream must be copied, therefore get the
stream
             StreamCache cache = (StreamCache) msg.getBody();
-            StreamCache copied = cache.copy();
+            StreamCache copied = cache.copy(answer);
             if (copied != null) {
                 msg.setBody(copied);
             }

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/test/java/org/apache/camel/processor/WireTapStreamCachingTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/WireTapStreamCachingTest.java
b/camel-core/src/test/java/org/apache/camel/processor/WireTapStreamCachingTest.java
index 0a87c13..1db7307 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/WireTapStreamCachingTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/WireTapStreamCachingTest.java
@@ -17,6 +17,7 @@
 package org.apache.camel.processor;
 
 import java.io.StringReader;
+
 import javax.xml.transform.stream.StreamSource;
 
 import org.apache.camel.ContextTestSupport;
@@ -26,6 +27,7 @@ import org.apache.camel.Message;
 import org.apache.camel.Processor;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
 
 /**
  * @version 
@@ -51,7 +53,19 @@ public class WireTapStreamCachingTest extends ContextTestSupport {
 
         assertMockEndpointsSatisfied();
     }
+    
+    @Test
+    public void testSendingAMessageUsingWiretapShouldNotDeleteStreamFileBeforeAllExcangesAreComplete()
throws InterruptedException {
+
+        x.expectedMessageCount(1);
+        y.expectedMessageCount(1);
+        z.expectedMessageCount(1);
+
+        // the used file should contain more than one character in order to be streamed into
the file system
+        template.sendBody("direct:a", this.getClass().getClassLoader().getResourceAsStream("org/apache/camel/processor/twoCharacters.txt"));
 
+        assertMockEndpointsSatisfied();
+    }
 
     @Override
     protected void setUp() throws Exception {
@@ -76,6 +90,8 @@ public class WireTapStreamCachingTest extends ContextTestSupport {
             public void configure() {
                 // enable stream caching
                 context.setStreamCaching(true);
+                // set stream threshold to 1, in order to stream into the file system
+                context.getStreamCachingStrategy().setSpoolThreshold(1);
 
                 errorHandler(deadLetterChannel("mock:error").redeliveryDelay(0).maximumRedeliveries(3));
 
@@ -83,7 +99,8 @@ public class WireTapStreamCachingTest extends ContextTestSupport {
                 from("direct:a").wireTap("direct:x").wireTap("direct:y").wireTap("direct:z");
 
                 from("direct:x").process(processor).to("mock:x");
-                from("direct:y").process(processor).to("mock:y");
+                // even if a process takes more time then the others the wire tap shall work
+                from("direct:y").delay(2000).process(processor).to("mock:y");
                 from("direct:z").process(processor).to("mock:z");
             }
         };

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/test/java/org/apache/camel/util/MessageHelperTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/util/MessageHelperTest.java b/camel-core/src/test/java/org/apache/camel/util/MessageHelperTest.java
index 7448989..cf80614 100644
--- a/camel-core/src/test/java/org/apache/camel/util/MessageHelperTest.java
+++ b/camel-core/src/test/java/org/apache/camel/util/MessageHelperTest.java
@@ -63,7 +63,7 @@ public class MessageHelperTest extends TestCase {
                 // noop
             }
 
-            public StreamCache copy() throws IOException {
+            public StreamCache copy(Exchange exchange) throws IOException {
                 return null;
             }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/camel-core/src/test/resources/org/apache/camel/processor/twoCharacters.txt
----------------------------------------------------------------------
diff --git a/camel-core/src/test/resources/org/apache/camel/processor/twoCharacters.txt b/camel-core/src/test/resources/org/apache/camel/processor/twoCharacters.txt
new file mode 100644
index 0000000..dfc9179
--- /dev/null
+++ b/camel-core/src/test/resources/org/apache/camel/processor/twoCharacters.txt
@@ -0,0 +1 @@
+AB
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/a99f6d57/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
----------------------------------------------------------------------
diff --git a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
index 8ecb8f7..b3afc4a 100644
--- a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
+++ b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
+import org.apache.camel.Exchange;
 import org.apache.camel.StreamCache;
 import org.apache.camel.util.IOHelper;
 import org.jboss.netty.buffer.ChannelBuffer;
@@ -87,7 +88,7 @@ public final class NettyChannelBufferStreamCache extends InputStream implements
     }
 
     @Override
-    public StreamCache copy() throws IOException {
+    public StreamCache copy(Exchange exchange) throws IOException {
         return new NettyChannelBufferStreamCache(buffer.copy());
     }
 


Mime
View raw message