/*
 * The MIT License
 *
 *  Copyright (c) 2015, Mahmoud Ben Hassine (mahmoud@benhassine.fr)
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a copy
 *  of this software and associated documentation files (the "Software"), to deal
 *  in the Software without restriction, including without limitation the rights
 *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *  copies of the Software, and to permit persons to whom the Software is
 *  furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *  THE SOFTWARE.
 */

package org.easybatch.flatfile;

import org.easybatch.core.api.Header;
import org.easybatch.core.api.RecordReader;
import org.easybatch.core.record.StringRecord;

import java.io.File;
import java.io.FileNotFoundException;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.Scanner;

/**
 * A {@link RecordReader} implementation that read data from a flat file.
 * <p/>
 * This reader produces {@link StringRecord} instances.
 *
 * @author Mahmoud Ben Hassine (mahmoud@benhassine.fr)
 */
public class FlatFileRecordReader implements RecordReader {

    /**
     * The current read record number.
     */
    private long currentRecordNumber;

    /**
     * The input file.
     */
    private File input;

    /**
     * The input file encoding name.
     */
    private String charsetName;

    /**
     * Scanner to read input file.
     */
    private Scanner scanner;

    /**
     * A second scanner used to calculate the number of records in the input file.
     * The main scanner may be used instead but since the {@link Scanner} class does not have a method to rewind it to the
     * beginning of the file ( {@link java.util.Scanner#reset()} does not rewind the scanner), another scanner instance is needed.
     */
    private Scanner recordCounterScanner;

    /**
     * Constructs a flat file record reader.
     *
     * @param input the input file
     * @throws FileNotFoundException thrown if the file does not exist
     */
    public FlatFileRecordReader(final File input) throws FileNotFoundException {
        this(input, Charset.defaultCharset().name());
    }

    /**
     * Constructs a flat file record reader.
     *
     * @param input       the input file
     * @param charsetName the encoding to use to read the file
     * @throws FileNotFoundException thrown if the file does not exist
     */
    public FlatFileRecordReader(final File input, final String charsetName) throws FileNotFoundException {
        this.input = input;
        this.charsetName = charsetName;

    }

    /**
     * {@inheritDoc}
     */
    public StringRecord readNextRecord() {
        Header header = new Header(++currentRecordNumber, getDataSourceName(), new Date());
        return new StringRecord(header, scanner.nextLine());
    }

    /**
     * {@inheritDoc}
     */
    public Long getTotalRecords() {
        long totalRecords = 0;
        while (recordCounterScanner.hasNextLine()) {
            totalRecords++;
            recordCounterScanner.nextLine();
        }
        recordCounterScanner.close();
        return totalRecords;
    }

    @Override
    public String getDataSourceName() {
        return input.getAbsolutePath();
    }

    /**
     * {@inheritDoc}
     */
    public void open() throws Exception {
        currentRecordNumber = 0;
        scanner = new Scanner(input, charsetName);
        recordCounterScanner = new Scanner(input);
    }

    /**
     * {@inheritDoc}
     */
    public boolean hasNextRecord() {
        return scanner.hasNextLine();
    }

    /**
     * {@inheritDoc}
     */
    public void close() {
        scanner.close();
    }

}

@@ -1365,16 +1365,48 @@
 o.File;%0A
+import java.io.FileInputStream;%0A
 import j
@@ -1518,16 +1518,56 @@
 anner;%0A%0A
+import static java.lang.String.format;%0A%0A
 /**%0A * A
@@ -3626,24 +3626,131 @@
 ecords = 0;%0A
+        try %7B%0A            recordCounterScanner = new Scanner(new FileInputStream(input), charsetName);%0A    
         whil
@@ -3802,16 +3802,20 @@
         
+    
 totalRec
@@ -3826,32 +3826,36 @@
 ++;%0A            
+  
+  
 recordCounterSca
@@ -3879,18 +3879,169 @@
         
+    
 %7D%0A
+        %7D catch (FileNotFoundException e) %7B%0A            throw new RuntimeException(format(%22File %25s not found%22, input), e);%0A        %7D finally %7B%0A    
         
@@ -4066,24 +4066,34 @@
 er.close();%0A
+        %7D%0A
         retu
@@ -4363,77 +4363,47 @@
 ner(
-input, charsetName);%0A        recordCounterScanner = new Scanner(input
+new FileInputStream(input), charsetName
 );%0A 

fix issue #53: incorrect record number with large data set