package com.flytxt.parser.processor;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.OverlappingFileLockException;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.Callable;
import java.util.regex.Pattern;

import javax.annotation.PreDestroy;

import lombok.Getter;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import com.flytxt.parser.marker.LineProcessor;
import com.flytxt.parser.marker.MarkerFactory;

@Component
@Scope("prototype")
public class FlyReader implements Callable<FlyReader> {

    private String folder;

    private LineProcessor lp;

    private boolean stopRequested;

    public enum Status {
        RUNNING, TERMINATED, SHUTTINGDOWN
    }

    @Getter
    private Status status;

    byte[] eol = System.lineSeparator().getBytes();

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public void set(final String folder, final LineProcessor lp) {
        this.lp = lp;
        this.folder = folder;
        logger.debug("file reader @ " + folder);
    }

    public void run() {
        final Path folderP = Paths.get(folder);
        if (!Files.exists(folderP)) {
            try {
                Files.createDirectories(folderP);
            } catch (final IOException e1) {
                logger.info("could not create input folder, stopping this FlyReader ", e1);
                stopRequested = true;
            }
        }
        logger.debug("Starting file reader @ " + folder);
        final ByteBuffer buf = ByteBuffer.allocate(6024);
        final MarkerFactory mf = new MarkerFactory();
        mf.setMaxListSize(lp.getMaxListSize());
        while (!stopRequested) {
            try (DirectoryStream<Path> directoryStream = Files.newDirectoryStream(Paths.get(folder))) {
                for (final Path path : directoryStream) {
                    final RandomAccessFile file = new RandomAccessFile(path.toString(), "rw");
                    logger.debug("picked up " + path.toString());
                    try {
                        lp.setInputFileName(path.getFileName().toString());
                        processFile(buf, path, file.getChannel(), mf);
                        buf.clear();
                        if (stopRequested) {
                            logger.debug("shutting down Worker @ :" + folder);
                            break;
                        }
                    } catch (final OverlappingFileLockException e) {
                        logger.error("Could not process " + path.toString(), e);
                    } finally {
                        file.close();
                    }
                }
            } catch (final Exception ex) {
                ex.printStackTrace();
            }
        }
        logger.debug("Worker down " + folder);
    }

    private void processFile(final ByteBuffer buf, final Path path, final FileChannel file, final MarkerFactory mf) throws IOException {
        final long t1 = System.nanoTime();
        readLines(file, buf, mf);
        lp.done();
        file.close();
        Files.delete(path);
        logger.debug("total time taken: " + (System.nanoTime() - t1));
        mf.printStat();
    }

    private final void readLines(final FileChannel file, final ByteBuffer buf, final MarkerFactory mf) throws IOException {
        int readCnt;
        final byte[] data = buf.array();
        buf.position(0);
        while ((readCnt = file.read(buf)) > 0) {
            long eolPosition;
            long previousEolPosition = 0;
            {
                do {
                    eolPosition = getEOLPosition(data, previousEolPosition + eol.length);
                    if (eolPosition < 0) {
                        if (previousEolPosition == 0) {
                            logger.debug("Increase byte array size");
                            System.exit(0); // or a better piece of code to exit the application
                        }
                        readLines(file.position(file.position()), (ByteBuffer) buf.limit((int) previousEolPosition + eol.length), mf); // total MESS
                        continue;
                    } else {
                        try {
                            final long T1 = System.nanoTime();
                            lp.process(data, (int) previousEolPosition, (int) eolPosition, mf);
                            logger.debug("Total: " + (System.nanoTime() - T1));
                            mf.reclaim();
                            previousEolPosition = eolPosition;
                        } catch (final IndexOutOfBoundsException e) {
                            logger.debug("could not process : " + new String(data, 0, (int) eolPosition) + " \n cause:" + e.getMessage());
                        }
                    }
                } while (eolPosition > 0);
            }
        }
    }

    @PreDestroy
    public void stop() {
        stopRequested = true;
    }

    @Override
    public FlyReader call() throws Exception {
        run();
        return this;
    }

    public boolean canProcess(final String folderName, final String fileName) {
        logger.debug("check " + folderName + " & " + lp.getFolder());
        if (lp.getFolder().equals(folderName)) {
            final String regex = lp.getFilter();
            if (regex == null) {
                return true;
            }
            final Pattern pattern = Pattern.compile(regex);
            return pattern.matcher(fileName).find();
        } else {
            return false;
        }
    }

    public long getEOLPosition(final byte[] data, final long startIndex) {
        try {
            int tokenIndex, currentIndex = (int) startIndex;
            while (currentIndex - startIndex <= data.length) {
                for (tokenIndex = 0; tokenIndex < eol.length && eol[tokenIndex] == data[currentIndex + tokenIndex]; tokenIndex++) { // loop to check if EOL is present at position currentIndex
                    ;
                }
                if (tokenIndex == eol.length) {
                    return currentIndex;
                }
                currentIndex++;
            }
        } catch (final Exception e) {
        }
        return -1;
    }
}
@@ -3251,32 +3251,41 @@
 t1 = System.
-nanoTime
+currentTimeMillis
 ();%0A        
@@ -3427,32 +3427,41 @@
 %22 + (System.
-nanoTime
+currentTimeMillis
 () - t1));%0A 
@@ -3680,33 +3680,8 @@
 ();%0A
-        buf.position(0);%0A
     
@@ -3886,16 +3886,22 @@
 on(data,
+ (int)
  previou
@@ -3925,16 +3925,25 @@
 l.length
+, readCnt
 );%0A     
@@ -4296,39 +4296,21 @@
 on()
-), (ByteBuffer) buf.limit((int)
+ - (readCnt -
  pre
@@ -4318,33 +4318,33 @@
 iousEolPosition 
-+
+-
  eol.length), mf
@@ -4343,28 +4343,41 @@
 gth)
+)
 , 
-mf); // total MESS
+(ByteBuffer) buf.clear(), mf);
 %0A   
@@ -4489,24 +4489,27 @@
             
+ //
  final long 
@@ -4572,24 +4572,55 @@
 rocess(data,
+ previousEolPosition == 0 ? 0 :
  (int) previ
@@ -4633,16 +4633,29 @@
 Position
+ + eol.length
 , (int) 
@@ -4691,32 +4691,35 @@
                 
+ //
  logger.debug(%22T
@@ -5926,20 +5926,19 @@
 , final 
-long
+int
  startIn
@@ -5940,16 +5940,36 @@
 artIndex
+, final int endIndex
 ) %7B%0A    
@@ -6020,22 +6020,16 @@
 tIndex =
- (int)
  startIn
@@ -6069,35 +6069,19 @@
 dex 
-- startIndex %3C= data.length
+%3C= endIndex
 ) %7B%0A

Fix in FlyReader#readLines