Allow to configure list of paths to upload

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

@@ -29,6 +29,7 @@ import com.google.common.base.Optional;
 import org.slf4j.Logger;
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
+import org.embulk.config.ConfigException;
 import org.embulk.config.ConfigInject;
 import org.embulk.config.ConfigSource;
 import org.embulk.config.ConfigDiff;
@@ -108,7 +109,12 @@ public class FilenameFileInputPlugin implements FileInputPlugin
     public interface PluginTask extends Task
     {
         @Config("path_prefix")
-        String getPathPrefix();
+        @ConfigDefault("null")
+        Optional <String> getPathPrefix();
+
+        @Config("path_list")
+        @ConfigDefault("null")
+        Optional <List<String>> getPathList();
 
         @Config("file_name_suffix")
         @ConfigDefault("null")
@@ -164,6 +170,14 @@ public class FilenameFileInputPlugin implements FileInputPlugin
     {
         PluginTask task = config.loadConfig(PluginTask.class);
 
+        // validate path_prefix: and path_list:
+        if (task.getPathPrefix().isPresent() && task.getPathList().isPresent()) {
+            throw new ConfigException("path_prefix: and path_list: must not be multi-select");
+        }
+        if (!task.getPathPrefix().isPresent() && !task.getPathList().isPresent()) {
+            throw new ConfigException("Must require path_prefix: or path_list:");
+        }
+
         // list files recursively
         List<String> files = new ArrayList<String>(listFiles(task));
 
@@ -279,7 +293,20 @@ public class FilenameFileInputPlugin implements FileInputPlugin
 
     public List<String> listFiles(PluginTask task)
     {
-        Path pathPrefix = Paths.get(task.getPathPrefix()).normalize();
+        final ImmutableList.Builder<String> builder = ImmutableList.builder();
+
+        if (task.getPathList().isPresent()) {
+          Path p;
+          for (String path_name : task.getPathList().get()) {
+              p = Paths.get(path_name).normalize();
+              if (Files.exists(p)) {
+                  builder.add(p.toString());
+              }
+          }
+        return builder.build();
+        } 
+
+        Path pathPrefix = Paths.get(task.getPathPrefix().orNull()).normalize();
         final Path directory;
         final String fileNamePrefix;
 
@@ -291,8 +318,6 @@ public class FilenameFileInputPlugin implements FileInputPlugin
             Path d = pathPrefix.getParent();
             directory = (d == null ? CURRENT_DIR : d);
         }
-
-        final ImmutableList.Builder<String> builder = ImmutableList.builder();
         
         final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss zzz");
         final String startDateString = task.getStartDate().orNull();
@@ -312,29 +337,24 @@ public class FilenameFileInputPlugin implements FileInputPlugin
         final String fileNameSuffix = task.getFileNameSuffix().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);
+        final boolean useLastModified = task.getUseLastModified();
+        if (useLastModified) { 
+            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;
             }
         } else {
-          lastModified = null;
+            lastModified = null;
         }
-        final boolean useLastModified = task.getUseLastModified();
-
-/*	      final Path yyy =  Paths.get("/mic/L0444-001/syscom/syscom004-14360007/events/2017/12/04/17338004.XMR");
-	         try {
-	          FileTime t = (FileTime) Files.getAttribute(yyy, "unix:ctime");
-		  log.info("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!time for {} = {}", yyy, t.toMillis());	
-         } catch (IOException e) {
-             log.info("Cannot get the unix:ctime - {} ", e);
-         }
-*/
         try {
             log.info("Listing local files at directory '{}' filtering filename by prefix '{}'", directory.equals(CURRENT_DIR) ? "." : directory.toString(), fileNamePrefix);
 
@@ -464,7 +484,7 @@ public class FilenameFileInputPlugin implements FileInputPlugin
         } catch (IOException ex) {
             throw new RuntimeException(String.format("Failed get a list of local files at '%s'", directory), ex);
         }
-      return builder.build();
+        return builder.build();
     }
 
     @Override