Merge pull request #8 from sakama/support-mapreduce-executor

Support MapReduce executor

Author: civitaspo

README.md

@@ -17,7 +17,7 @@ Stores files on a SFTP Server
 - **port**: (string, default: `22`)
 - **user**: (string, required)
 - **password**: (string, default: `null`)
-- **secret_key_file**: (string, default: `null`)
+- **secret_key_file**: (string, default: `null`) [see below](#secret-keyfile-configuration)
 - **secret_key_passphrase**: (string, default: `""`)
 - **user_directory_is_root**: (boolean, default: `true`)
 - **timeout**: sftp connection timeout seconds (integer, default: `600`)
@@ -42,6 +42,32 @@ out:
   sequence_format: ".%01d%01d"
 ```
 
+### Secret Keyfile configuration
+
+Please set path of secret_key_file as follows.
+```
+out:
+  type: sftp
+  ...
+  secret_key_file: /path/to/id_rsa
+  ...
+```
+
+You can also embed contents of secret_key_file at config.yml.
+```
+out:
+  type: sftp
+  ...
+  secret_key_file:
+    content |
+      -----BEGIN RSA PRIVATE KEY-----
+      ABCDEFG...
+      HIJKLMN...
+      OPQRSTU...
+      -----END RSA PRIVATE KEY-----
+  ...
+```
+
 ## Run Example
 replace settings in `example/sample.yml` before running.
 

src/main/java/org/embulk/output/sftp/SftpFileOutput.java

@@ -1,5 +1,6 @@
 package org.embulk.output.sftp;
 
+import com.google.common.base.Function;
 import com.google.common.base.Throwables;
 import org.apache.commons.vfs2.FileObject;
 import org.apache.commons.vfs2.FileSystemException;
@@ -12,6 +13,7 @@
 import org.embulk.spi.Exec;
 import org.embulk.spi.FileOutput;
 import org.embulk.spi.TransactionalFileOutput;
+import org.embulk.spi.unit.LocalFile;
 import org.slf4j.Logger;
 
 import java.io.File;
@@ -80,7 +82,10 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
             SftpFileSystemConfigBuilder.getInstance().setTimeout(fsOptions, task.getSftpConnectionTimeout());
             SftpFileSystemConfigBuilder.getInstance().setStrictHostKeyChecking(fsOptions, "no");
             if (task.getSecretKeyFilePath().isPresent()) {
-                IdentityInfo identityInfo = new IdentityInfo(new File((task.getSecretKeyFilePath().get())), task.getSecretKeyPassphrase().getBytes());
+                IdentityInfo identityInfo = new IdentityInfo(
+                    new File((task.getSecretKeyFilePath().transform(localFileToPathString()).get())),
+                    task.getSecretKeyPassphrase().getBytes()
+                );
                 SftpFileSystemConfigBuilder.getInstance().setIdentityInfo(fsOptions, identityInfo);
                 logger.info("set identity: {}", task.getSecretKeyFilePath().get());
             }
@@ -230,4 +235,15 @@ private FileObject newSftpFile(URI sftpUri)
             }
         }
     }
+
+    private Function<LocalFile, String> localFileToPathString()
+    {
+        return new Function<LocalFile, String>()
+        {
+            public String apply(LocalFile file)
+            {
+                return file.getPath().toString();
+            }
+        };
+    }
 }

src/main/java/org/embulk/output/sftp/SftpFileOutputPlugin.java

@@ -11,6 +11,7 @@
 import org.embulk.spi.Exec;
 import org.embulk.spi.FileOutputPlugin;
 import org.embulk.spi.TransactionalFileOutput;
+import org.embulk.spi.unit.LocalFile;
 import org.slf4j.Logger;
 
 import java.util.List;
@@ -39,7 +40,8 @@ public interface PluginTask
 
         @Config("secret_key_file")
         @ConfigDefault("null")
-        public Optional<String> getSecretKeyFilePath();
+        public Optional<LocalFile> getSecretKeyFilePath();
+        public void setSecretKeyFilePath(Optional<LocalFile> secretKeyFilePath);
 
         @Config("secret_key_passphrase")
         @ConfigDefault("\"\"")