Merge pull request #4 from civitaspo/test

Add Test

Author: civitaspo

.travis.yml

@@ -0,0 +1,9 @@
+language: java
+jdk:
+  - openjdk7
+  - oraclejdk7
+  - oraclejdk8
+script:
+  - ./gradlew test
+after_success:
+  - ./gradlew jacocoTestReport coveralls

README.md

@@ -1,4 +1,6 @@
 # Sftp file output plugin for Embulk
+[![Build Status](https://travis-ci.org/civitaspo/embulk-output-sftp.svg)](https://travis-ci.org/civitaspo/embulk-output-sftp)
+[![Coverage Status](https://coveralls.io/repos/civitaspo/embulk-output-sftp/badge.svg?branch=master&service=github)](https://coveralls.io/github/civitaspo/embulk-output-sftp?branch=master)
 
 Stores files on a SFTP Server
 

build.gradle

@@ -1,6 +1,8 @@
 plugins {
     id "com.jfrog.bintray" version "1.1"
     id "com.github.jruby-gradle.base" version "0.1.5"
+    id "com.github.kt3k.coveralls" version "2.4.0"
+    id "jacoco"
     id "java"
 }
 import com.github.jrubygradle.JRubyExec
@@ -17,12 +19,22 @@ sourceCompatibility = 1.7
 targetCompatibility = 1.7
 
 dependencies {
-    compile  "org.embulk:embulk-core:0.7.4"
-    provided "org.embulk:embulk-core:0.7.4"
+    compile  "org.embulk:embulk-core:0.7.+"
+    provided "org.embulk:embulk-core:0.7.+"
     // compile "YOUR_JAR_DEPENDENCY_GROUP:YOUR_JAR_DEPENDENCY_MODULE:YOUR_JAR_DEPENDENCY_VERSION"
     compile "org.apache.commons:commons-vfs2:2.+"
     compile "com.jcraft:jsch:0.1.53"
     testCompile "junit:junit:4.+"
+    testCompile "org.embulk:embulk-core:0.7.+:tests"
+    testCompile "org.embulk:embulk-standards:0.7.+"
+    testCompile "org.apache.sshd:apache-sshd:1.+"
+}
+
+jacocoTestReport {
+    reports {
+        xml.enabled = true // coveralls plugin depends on xml format report
+        html.enabled = true
+    }
 }
 
 task classpath(type: Copy, dependsOn: ["jar"]) {

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

@@ -4,7 +4,6 @@
 import org.apache.commons.vfs2.FileObject;
 import org.apache.commons.vfs2.FileSystemException;
 import org.apache.commons.vfs2.FileSystemOptions;
-import org.apache.commons.vfs2.Selectors;
 import org.apache.commons.vfs2.impl.StandardFileSystemManager;
 import org.apache.commons.vfs2.provider.sftp.IdentityInfo;
 import org.apache.commons.vfs2.provider.sftp.SftpFileSystemConfigBuilder;
@@ -35,6 +34,7 @@ public class SftpFileOutput
     private final String userInfo;
     private final String host;
     private final int port;
+    private final int maxConnectionRetry;
     private final String pathPrefix;
     private final String sequenceFormat;
     private final String fileNameExtension;
@@ -82,6 +82,7 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
             if (task.getSecretKeyFilePath().isPresent()) {
                 IdentityInfo identityInfo = new IdentityInfo(new File((task.getSecretKeyFilePath().get())), task.getSecretKeyPassphrase().getBytes());
                 SftpFileSystemConfigBuilder.getInstance().setIdentityInfo(fsOptions, identityInfo);
+                logger.info("set identity: {}", task.getSecretKeyFilePath().get());
             }
         }
         catch (FileSystemException e) {
@@ -99,6 +100,7 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
         this.fsOptions = initializeFsOptions(task);
         this.host = task.getHost();
         this.port = task.getPort();
+        this.maxConnectionRetry = task.getMaxConnectionRetry();
         this.pathPrefix = task.getPathPrefix();
         this.sequenceFormat = task.getSequenceFormat();
         this.fileNameExtension = task.getFileNameExtension();
@@ -204,6 +206,26 @@ private String getOutputFilePath()
     private FileObject newSftpFile(URI sftpUri)
             throws FileSystemException
     {
-        return manager.resolveFile(sftpUri.toString(), fsOptions);
+        int count = 0;
+        while (true) {
+            try {
+                return manager.resolveFile(sftpUri.toString(), fsOptions);
+            }
+            catch (FileSystemException e) {
+                if (++count == maxConnectionRetry) {
+                    throw e;
+                }
+                logger.warn("failed to connect sftp server: " + e.getMessage(), e);
+
+                try {
+                    Thread.sleep(count * 1000); // milliseconds
+                }
+                catch (InterruptedException e1) {
+                    // Ignore this exception
+                    logger.warn(e.getMessage(), e);
+                }
+                logger.warn("retry to connect sftp server: " + count + " times");
+            }
+        }
     }
 }

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

@@ -53,6 +53,10 @@ public interface PluginTask
         @ConfigDefault("600") // 10 minutes
         public int getSftpConnectionTimeout();
 
+        @Config("max_connection_retry")
+        @ConfigDefault("5") // 5 times retry to connect sftp server if failed.
+        public int getMaxConnectionRetry();
+
         @Config("path_prefix")
         public String getPathPrefix();