remove working file and upload directly

Author: civitaspo

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

@@ -35,15 +35,14 @@ public class SftpFileOutput
     private final String userInfo;
     private final String host;
     private final int port;
-    private final String workingFileScheme;
     private final String pathPrefix;
     private final String sequenceFormat;
     private final String fileNameExtension;
 
     private final int taskIndex;
     private int fileIndex = 0;
-    private FileObject currentWorkingFile;
-    private OutputStream currentWorkingFileOutputStream;
+    private FileObject currentFile;
+    private OutputStream currentFileOutputStream;
 
     private StandardFileSystemManager initializeStandardFileSystemManager()
     {
@@ -100,7 +99,6 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
         this.host = task.getHost();
         this.port = task.getPort();
         this.pathPrefix = task.getPathPrefix();
-        this.workingFileScheme = task.getWorkingFileScheme();
         this.sequenceFormat = task.getSequenceFormat();
         this.fileNameExtension = task.getFileNameExtension();
         this.taskIndex = taskIndex;
@@ -109,12 +107,12 @@ private FileSystemOptions initializeFsOptions(PluginTask task)
     @Override
     public void nextFile()
     {
-        closeCurrentWithUpload();
+        closeCurrentFile();
 
         try {
-            currentWorkingFile = newWorkingFile(getWorkingFileUri(getOutputFilePath()));
-            currentWorkingFileOutputStream = currentWorkingFile.getContent().getOutputStream();
-            logger.info("new working file: {}", currentWorkingFile.getPublicURIString());
+            currentFile = newSftpFile(getSftpFileUri(getOutputFilePath()));
+            currentFileOutputStream = currentFile.getContent().getOutputStream();
+            logger.info("new sftp file: {}", currentFile.getPublicURIString());
         }
         catch (FileSystemException e) {
             logger.error(e.getMessage());
@@ -129,12 +127,12 @@ public void nextFile()
     @Override
     public void add(Buffer buffer)
     {
-        if (currentWorkingFile == null) {
+        if (currentFile == null) {
             throw new IllegalStateException("nextFile() must be called before poll()");
         }
 
         try {
-            currentWorkingFileOutputStream.write(buffer.array(), buffer.offset(), buffer.limit());
+            currentFileOutputStream.write(buffer.array(), buffer.offset(), buffer.limit());
         }
         catch (IOException e) {
             logger.error(e.getMessage());
@@ -146,13 +144,13 @@ public void add(Buffer buffer)
     @Override
     public void finish()
     {
-        closeCurrentWithUpload();
+        closeCurrentFile();
     }
 
     @Override
     public void close()
     {
-        closeCurrentWithUpload();
+        closeCurrentFile();
         manager.close();
     }
 
@@ -167,57 +165,24 @@ public TaskReport commit()
         return null;
     }
 
-    private void closeCurrentWithUpload()
+
+    private void closeCurrentFile()
     {
-        try {
-            closeCurrentWorkingFileContent();
-            uploadCurrentWorkingFileToSftp();
-            closeCurrentWorkingFile();
+        if (currentFile == null) {
+            return;
         }
-        catch (URISyntaxException e) {
-            logger.error(e.getMessage());
-            Throwables.propagate(e);
+
+        try {
+            currentFileOutputStream.close();
+            currentFile.getContent().close();
+            currentFile.close();
         }
         catch (IOException e) {
             logger.error(e.getMessage());
             Throwables.propagate(e);
         }
         fileIndex++;
-        currentWorkingFile = null;
-    }
-
-    private void closeCurrentWorkingFileContent()
-            throws IOException
-    {
-        if (currentWorkingFile == null) {
-            return;
-        }
-        currentWorkingFileOutputStream.close();
-        currentWorkingFile.getContent().close();
-    }
-
-    private void uploadCurrentWorkingFileToSftp()
-            throws FileSystemException, URISyntaxException
-    {
-        if (currentWorkingFile == null) {
-            return;
-        }
-
-        try (FileObject remoteSftpFile = newSftpFile(getSftpFileUri(getOutputFilePath()))) {
-            remoteSftpFile.copyFrom(currentWorkingFile, Selectors.SELECT_SELF);
-            logger.info("Upload: {}", remoteSftpFile.getPublicURIString());
-        }
-    }
-
-    private void closeCurrentWorkingFile()
-            throws FileSystemException
-    {
-        if (currentWorkingFile == null) {
-            return;
-        }
-
-        currentWorkingFile.close();
-        currentWorkingFile.delete();
+        currentFile = null;
     }
 
     private URI getSftpFileUri(String remoteFilePath)
@@ -226,12 +191,6 @@ private URI getSftpFileUri(String remoteFilePath)
         return new URI("sftp", userInfo, host, port, remoteFilePath, null, null);
     }
 
-    private URI getWorkingFileUri(String remoteFilePath)
-            throws URISyntaxException
-    {
-        return new URI(workingFileScheme, null, remoteFilePath, null);
-    }
-
     private String getOutputFilePath()
     {
         return pathPrefix + String.format(sequenceFormat, taskIndex, fileIndex) + fileNameExtension;
@@ -242,12 +201,4 @@ private FileObject newSftpFile(URI sftpUri)
     {
         return manager.resolveFile(sftpUri.toString(), fsOptions);
     }
-
-    private FileObject newWorkingFile(URI workingFileUri)
-            throws FileSystemException
-    {
-        FileObject workingFile = manager.resolveFile(workingFileUri);
-        workingFile.createFile();
-        return workingFile;
-    }
 }

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

@@ -53,10 +53,6 @@ public interface PluginTask
         @ConfigDefault("600") // 10 munites
         public int getSftpConnectionTimeout();
 
-        @Config("working_file_schema")
-        @ConfigDefault("ram")
-        public String getWorkingFileScheme();
-
         @Config("path_prefix")
         public String getPathPrefix();