Merge pull request #50 from tvhung83/add_local_buffering_config

Add `local_buffering` & `temp_file_threshold` configs

Author: tvhung83

.travis.yml

@@ -1,8 +1,7 @@
-dist: precise
+dist: trusty
 language: java
 jdk:
-  - openjdk7
-  - oraclejdk7
+  - openjdk8
   - oraclejdk8
 script:
   - ./gradlew test

build.gradle

@@ -22,18 +22,19 @@ sourceCompatibility = 1.7
 targetCompatibility = 1.7
 
 dependencies {
-    compile  "org.embulk:embulk-core:0.8.6"
-    provided "org.embulk:embulk-core:0.8.6"
+    compile  "org.embulk:embulk-core:0.9.7"
+    provided "org.embulk:embulk-core:0.9.7"
     // compile "YOUR_JAR_DEPENDENCY_GROUP:YOUR_JAR_DEPENDENCY_MODULE:YOUR_JAR_DEPENDENCY_VERSION"
     compile "org.apache.commons:commons-vfs2:2.2"
     compile "commons-io:commons-io:2.6"
     compile "com.jcraft:jsch:0.1.54"
     testCompile "junit:junit:4.+"
-    testCompile "org.embulk:embulk-core:0.8.6:tests"
-    testCompile "org.embulk:embulk-standards:0.8.6"
+    testCompile "org.embulk:embulk-core:0.9.7:tests"
+    testCompile "org.embulk:embulk-standards:0.9.7"
     testCompile "org.apache.sshd:apache-sshd:1.1.0+"
     testCompile "org.littleshoot:littleproxy:1.1.0-beta1"
     testCompile "io.netty:netty-all:4.0.34.Final"
+    testCompile "org.mockito:mockito-core:2.+"
 }
 
 jacocoTestReport {

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

@@ -13,6 +13,9 @@
 import org.embulk.spi.TransactionalFileOutput;
 import org.embulk.spi.unit.LocalFile;
 
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
 import java.util.List;
 import java.util.Map;
 
@@ -47,7 +50,7 @@ public interface PluginTask
 
         @Config("user_directory_is_root")
         @ConfigDefault("true")
-        public Boolean getUserDirIsRoot();
+        public boolean getUserDirIsRoot();
 
         @Config("timeout")
         @ConfigDefault("600") // 10 minutes
@@ -73,7 +76,18 @@ public interface PluginTask
 
         @Config("rename_file_after_upload")
         @ConfigDefault("false")
-        public Boolean getRenameFileAfterUpload();
+        public boolean getRenameFileAfterUpload();
+
+        // if `false`, plugin will use remote file as buffer
+        @Config("local_buffering")
+        @ConfigDefault("true")
+        public boolean getLocalBuffering();
+
+        @Min(50L * 1024 * 1024) // 50MiB
+        @Max(10L * 1024 * 1024 * 1024) // 10GiB
+        @Config("temp_file_threshold")
+        @ConfigDefault("5368709120") // 5GiB
+        public long getTempFileThreshold();
     }
 
     @Override
@@ -141,6 +155,9 @@ public void cleanup(TaskSource taskSource,
     public TransactionalFileOutput open(TaskSource taskSource, final int taskIndex)
     {
         final PluginTask task = taskSource.loadTask(PluginTask.class);
-        return new SftpFileOutput(task, taskIndex);
+        if (task.getLocalBuffering()) {
+            return new SftpLocalFileOutput(task, taskIndex);
+        }
+        return new SftpRemoteFileOutput(task, taskIndex);
     }
 }