start to fix the last path support

src/main/java/org/embulk/input/filename/FilenameInputPlugin.java

@@ -57,9 +57,9 @@ public class FilenameInputPlugin
         @ConfigDefault("[]")
         ArrayList<String> getMultiTag();
         
-        @Config("last_path")
-        @ConfigDefault("null")
-        Optional<String> getLastPath();
+        @Config("last_paths")
+        @ConfigDefault("[]")
+        ArrayList<String> getLastPaths();
         
         @Config("order_by_modified_time")
         @ConfigDefault("0")
@@ -98,7 +98,7 @@ public class FilenameInputPlugin
     
     private static int chunkSize;
     
-    private static ArrayList<String> last_p = new ArrayList<String>();
+    private static ArrayList<String> lastPaths = new ArrayList<String>();
 
     @Override
     public ConfigDiff transaction(ConfigSource config,
@@ -117,14 +117,19 @@ public class FilenameInputPlugin
                 // If the Number of tags is less than the directories, we say that the default tag is ""
                 tagList.add("");
             }
+            while (lastPaths.size()< dirList.size()){
+                lastPaths.add("");
+            }
         } else {
             throw new RuntimeException("The multi_dir should contain at least 1 directory.");
         }
         
         
 
-        for ( String dir : dirList){
-            ArrayList<String> files = listFiles(task,Paths.get(dir).normalize());
+        for (int i =0; i<dirList.size(); i++ ){
+            String dir = dirList.get(i);
+            String last_path = lastPaths.get(i);
+            ArrayList<String> files = listFiles(task,Paths.get(dir).normalize(),last_path);
             // Sort the files if each directory
             int order_modified = task.getOrderByModifiedTime();
             int order_creation = task.getOrderByCreationTime();
@@ -279,7 +284,7 @@ public class FilenameInputPlugin
     }
     
     
-    public ArrayList<String> listFiles(PluginTask task,Path pathPrefix)
+    public ArrayList<String> listFiles(PluginTask task,Path pathPrefix, String lastPath)
     {
         //Path pathPrefix = Paths.get(task.getPathPrefix()).normalize();
         final Path directory;
@@ -293,20 +298,27 @@ public class FilenameInputPlugin
             directory = (d == null ? CURRENT_DIR : d);
         }
 
-        //final ImmutableList.Builder<String> builder = ImmutableList.builder();
         final ArrayList<String> filesArray = new ArrayList<String>();
-        final String lastPath = task.getLastPath().orNull();
         final Integer fileSize = task.getFileSize().orNull();
         try {
             log.info("Listing local files at directory '{}' filtering filename by prefix '{}'", directory.equals(CURRENT_DIR) ? "." : directory.toString(), fileNamePrefix);
             Files.walkFileTree(directory, new SimpleFileVisitor<Path>() {
+                
+                // This method check the dirname
                 @Override
                 public FileVisitResult preVisitDirectory(Path path, BasicFileAttributes attrs)
                 {
                     if (path.equals(directory)) {
                         return FileVisitResult.CONTINUE;
-                    } else if (lastPath != null && path.toString().compareTo(lastPath.substring(0, path.toString().length())) < 0) {
+                    } else if (lastPath != "") {
+                        if ( order == 1 && path.toString().compareTo(lastPath.substring(0, path.toString().length())) < 0)
+                        {
                             return FileVisitResult.SKIP_SUBTREE;
+                        }
+                        else if (order ==2 && path.toString().compareTo(lastPath.substring(0, path.toString().length())) > 0)
+                        {
+                            return FileVisitResult.SKIP_SUBTREE;
+                        }
                     } else if (path.getFileName().toString().startsWith(".")) {
                         return FileVisitResult.SKIP_SUBTREE;
                     } else {
@@ -318,7 +330,7 @@ public class FilenameInputPlugin
                     }
                 }
 
-
+                // This method check the filename
                 @Override
                 public FileVisitResult visitFile(Path path, BasicFileAttributes attrs)
                 {