embulk-input-filename: get last_change_time from last_path if not defined. Use long.

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

@@ -145,7 +145,7 @@ public class FilenameFileInputPlugin implements FileInputPlugin
 
         @Config("last_modified")
         @ConfigDefault("null")
-        Optional<String> getLastModified();
+        Optional<Long> getLastModified();
 
 
         List<String> getFiles();
@@ -311,7 +311,20 @@ public class FilenameFileInputPlugin implements FileInputPlugin
         final Integer fileSize = task.getFileSize().orNull();
         final String fileNameSuffix = task.getFileNameSuffix().orNull();
 
-        final String lastModified =  task.getLastModified().orNull();
+        final Long lastModified;
+        if (task.getLastModified().orNull() != null) {
+            lastModified = task.getLastModified().orNull();
+        } else if (lastPath != null) {
+            FileTime lastPathTime;
+            try {
+                lastPathTime = (FileTime) Files.getAttribute(Paths.get(lastPath), "unix:ctime");
+                lastModified = lastPathTime.toMillis();
+            } catch (IOException e) {
+                throw new RuntimeException(String.format("Cannot get the last changed time of lastpath '%s'", lastPath), e);
+            }
+        } else {
+          lastModified = null;
+        }
         final boolean useLastModified = task.getUseLastModified();
 
 /*	      final Path yyy =  Paths.get("/mic/L0444-001/syscom/syscom004-14360007/events/2017/12/04/17338004.XMR");
@@ -346,7 +359,7 @@ public class FilenameFileInputPlugin implements FileInputPlugin
         	                    return FileVisitResult.CONTINUE;
                                 } else if (firstPath != null && path.toString().compareTo(firstPath.substring(0, path.toString().length())) < 0) {
                                     return FileVisitResult.SKIP_SUBTREE;
-                	        } else if (lastModified != null && Long.parseLong(lastModified) > fileChangeTime) { //if file is older then lastModified
+                	        } else if (lastModified != null && lastModified > fileChangeTime) { //if file is older then lastModified
 	                            return FileVisitResult.SKIP_SUBTREE;
         	                } else if (path.getFileName().toString().startsWith(".")) {
                 	            return FileVisitResult.SKIP_SUBTREE;
@@ -410,9 +423,9 @@ public class FilenameFileInputPlugin implements FileInputPlugin
 //			log.info ("file = {}", path);
 //                      log.info("last_Modified = {},    files time = {}", lastModified, fileChangeTime);
 
-                        if (lastModified != null && Long.parseLong(lastModified) > fileChangeTime) {
+                        if (lastModified != null && lastModified > fileChangeTime) {
                             return FileVisitResult.CONTINUE;
-                        } else if (lastModified != null && Long.parseLong(lastModified) == fileChangeTime && lastPath != null && path.toString().compareTo(lastPath) <= 0 ) {
+                        } else if (lastModified != null && lastModified == fileChangeTime && lastPath != null && path.toString().compareTo(lastPath) <= 0 ) {
 			    return FileVisitResult.CONTINUE;
 			}
 			  else if (path.getFileName().toString().startsWith(".")) {