Csv iterator CAMEL-7080

camel-core/src/main/java/org/apache/camel/builder/DataFormatClause.java

@@ -180,6 +180,14 @@ public T csv() {
         return dataFormat(new CsvDataFormat());
     }
 
+    /**
+     * Uses the CSV data format for a huge file.
+     * Sequential access through an iterator.
+     */
+    public T csvLazyLoad() {
+        return dataFormat(new CsvDataFormat(true));
+    }
+
     /**
      * Uses the custom data format
      */

camel-core/src/main/java/org/apache/camel/model/dataformat/CsvDataFormat.java

@@ -46,6 +46,8 @@ public class CsvDataFormat extends DataFormatDefinition {
     private String strategyRef;
     @XmlAttribute
     private Boolean skipFirstLine;
+    @XmlAttribute
+    private Boolean lazyLoad;
 
     public CsvDataFormat() {
         super("csv");
@@ -56,6 +58,11 @@ public CsvDataFormat(String delimiter) {
         setDelimiter(delimiter);
     }
 
+    public CsvDataFormat(boolean lazyLoad) {
+        this();
+        setLazyLoad(lazyLoad);
+    }
+
     public Boolean isAutogenColumns() {
         return autogenColumns;
     }
@@ -96,6 +103,14 @@ public void setSkipFirstLine(Boolean skipFirstLine) {
         this.skipFirstLine = skipFirstLine;
     }
 
+    public Boolean getLazyLoad() {
+        return lazyLoad;
+    }
+
+    public void setLazyLoad(Boolean lazyLoad) {
+        this.lazyLoad = lazyLoad;
+    }
+
     @Override
     protected DataFormat createDataFormat(RouteContext routeContext) {
         DataFormat csvFormat = super.createDataFormat(routeContext);
@@ -131,5 +146,9 @@ protected void configureDataFormat(DataFormat dataFormat, CamelContext camelCont
         if (skipFirstLine != null) {
             setProperty(camelContext, dataFormat, "skipFirstLine", skipFirstLine);
         }
+
+        if (lazyLoad != null) {
+            setProperty(camelContext, dataFormat, "lazyLoad", lazyLoad);
+        }
     }
-}
+}

components/camel-csv/pom.xml

@@ -56,6 +56,12 @@
       <artifactId>slf4j-log4j12</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.googlecode.jmockit</groupId>
+      <artifactId>jmockit</artifactId>
+      <version>1.5</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

components/camel-csv/src/main/java/org/apache/camel/dataformat/csv/CsvDataFormat.java

@@ -16,13 +16,15 @@
  */
 package org.apache.camel.dataformat.csv;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -54,6 +56,10 @@ public class CsvDataFormat implements DataFormat {
     private boolean autogenColumns = true;
     private String delimiter;
     private boolean skipFirstLine;
+    /**
+     * Lazy row loading with iterator for big files.
+     */
+    private boolean lazyLoad;
 
     public void marshal(Exchange exchange, Object object, OutputStream outputStream) throws Exception {
         if (delimiter != null) {
@@ -96,32 +102,42 @@ public Object unmarshal(Exchange exchange, InputStream inputStream) throws Excep
         strategy.setDelimiter(config.getDelimiter());
 
         InputStreamReader in = new InputStreamReader(inputStream, IOHelper.getCharsetName(exchange));
-
+        CsvIterator csvIterator;
         try {
-            CSVParser parser = new CSVParser(in, strategy);
-            List<List<String>> list = new ArrayList<List<String>>();
-            boolean isFirstLine = true;
-            while (true) {
-                String[] strings = parser.getLine();
-                if (isFirstLine) {
-                    isFirstLine = false;
-                    if (skipFirstLine) {
-                        // skip considering the first line if we're asked to do so
-                        continue;
-                    }
-                }
-                if (strings == null) {
-                    break;
-                }
-                List<String> line = Arrays.asList(strings);
-                list.add(line);
+            CSVParser parser = createParser(in);
+            if (parser == null) {
+                IOHelper.close(in);
+                return Collections.emptyIterator();
             }
-            return list;
-        } finally {
+            csvIterator = new CsvIterator(parser, in);
+        } catch (IOException e) {
             IOHelper.close(in);
+            throw e;
+        }
+        if (lazyLoad) {
+            return csvIterator;
         }
+        return loadAllAsList(csvIterator);
     }
-
+
+    private CSVParser createParser(InputStreamReader in) throws IOException {
+        CSVParser parser = new CSVParser(in, strategy);
+        if (skipFirstLine) {
+            if (null == parser.getLine()) {
+                return null;
+            }
+        }
+        return parser;
+    }
+
+    private List<List<String>> loadAllAsList(CsvIterator iter) throws IOException {
+        List<List<String>> list = new ArrayList<List<String>>();
+        while (iter.hasNext()) {
+            list.add(iter.next());
+        }
+        return list;
+    }
+
     public String getDelimiter() {
         return delimiter;
     }
@@ -170,6 +186,14 @@ public void setSkipFirstLine(boolean skipFirstLine) {
         this.skipFirstLine = skipFirstLine;
     }
 
+    public boolean isLazyLoad() {
+        return lazyLoad;
+    }
+
+    public void setLazyLoad(boolean lazyLoad) {
+        this.lazyLoad = lazyLoad;
+    }
+
     private synchronized void updateFieldsInConfig(Set<?> set, Exchange exchange) {
         for (Object value : set) {
             if (value != null) {

components/camel-csv/src/main/java/org/apache/camel/dataformat/csv/CsvIterator.java

@@ -0,0 +1,63 @@
+package org.apache.camel.dataformat.csv;
+
+import org.apache.camel.util.IOHelper;
+import org.apache.commons.csv.CSVParser;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/**
+ */
+public class CsvIterator implements Iterator<List<String>>, Closeable {
+
+    private final CSVParser parser;
+    private final InputStreamReader in;
+    private String[] line;
+
+    public CsvIterator(CSVParser parser, InputStreamReader in)
+            throws IOException
+    {
+        this.parser = parser;
+        this.in = in;
+        line = parser.getLine();
+    }
+
+    @Override
+    public boolean hasNext() {
+        return line != null;
+    }
+
+    @Override
+    public List<String> next() {
+        if (!hasNext()) {
+            throw new NoSuchElementException();
+        }
+        List<String> result = Arrays.asList(line);
+        try {
+            line = parser.getLine();
+        } catch (IOException e) {
+            line = null;
+            IOHelper.close(in);
+            throw new IllegalStateException(e);
+        }
+        if (line == null) {
+            IOHelper.close(in);
+        }
+        return result;
+    }
+
+    @Override
+    public void remove() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void close() throws IOException {
+        in.close();
+    }
+}

components/camel-csv/src/test/java/org/apache/camel/dataformat/csv/CsvIteratorTest.java

@@ -0,0 +1,100 @@
+package org.apache.camel.dataformat.csv;
+
+import mockit.Expectations;
+import mockit.Injectable;
+import org.apache.commons.csv.CSVParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.Arrays;
+import java.util.NoSuchElementException;
+
+/**
+ */
+public class CsvIteratorTest {
+
+    public static final String HDD_CRASH = "HDD crash";
+
+    @Test
+    public void closeIfError(
+            final @Injectable InputStreamReader reader,
+            final @Injectable CSVParser parser)
+            throws IOException
+    {
+        new Expectations() {
+            {
+                parser.getLine();
+                result = new String[] { "1" };
+
+                parser.getLine();
+                result = new String[] { "2" };
+
+                parser.getLine();
+                result = new IOException(HDD_CRASH);
+
+                reader.close();
+            }
+        };
+
+        CsvIterator iterator = new CsvIterator(parser, reader);
+        Assert.assertTrue(iterator.hasNext());
+        Assert.assertEquals(Arrays.asList("1"), iterator.next());
+        Assert.assertTrue(iterator.hasNext());
+
+        try {
+            iterator.next();
+            Assert.fail("exception expected");
+        } catch (IllegalStateException e) {
+            Assert.assertEquals(HDD_CRASH, e.getCause().getMessage());
+        }
+
+        Assert.assertFalse(iterator.hasNext());
+
+        try {
+            iterator.next();
+            Assert.fail("exception expected");
+        } catch (NoSuchElementException e) {
+            // okay
+        }
+    }
+
+    @Test
+    public void normalCycle(final @Injectable InputStreamReader reader,
+                            final @Injectable CSVParser parser)
+            throws IOException
+    {
+        new Expectations() {
+            {
+                parser.getLine();
+                result = new String[] { "1" };
+
+                parser.getLine();
+                result = new String[] { "2" };
+
+                parser.getLine();
+                result = null;
+
+                reader.close();
+            }
+        };
+
+        CsvIterator iterator = new CsvIterator(parser, reader);
+        Assert.assertTrue(iterator.hasNext());
+        Assert.assertEquals(Arrays.asList("1"), iterator.next());
+
+        Assert.assertTrue(iterator.hasNext());
+        Assert.assertEquals(Arrays.asList("2"), iterator.next());
+
+        Assert.assertFalse(iterator.hasNext());
+
+        try {
+            iterator.next();
+            Assert.fail("exception expected");
+        } catch (NoSuchElementException e) {
+            // okay
+        }
+
+    }
+}