[Improve][Connector-V2] Refactor local file sink connector code struc…

…ture (apache#2655) * [Improve][Connector-V2] Refactor local file code struct
laglangyue · Sep 11, 2022 · 856d0d7 · 856d0d7
1 parent 5db3e36
commit 856d0d7
Show file tree

Hide file tree

Showing 13 changed files with 26 additions and 1,290 deletions.
diff --git a/...ocal/sink/filesystem/LocalFileSystem.java → ...eatunnel/file/local/config/LocalConf.java b/...ocal/sink/filesystem/LocalFileSystem.java → ...eatunnel/file/local/config/LocalConf.java
@@ -15,29 +15,14 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.connectors.seatunnel.file.local.sink.filesystem;
+package org.apache.seatunnel.connectors.seatunnel.file.local.config;
 
-import org.apache.seatunnel.connectors.seatunnel.file.sink.spi.FileSystem;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
+public class LocalConf extends HadoopConf {
+    private final String fsHdfsImpl = "org.apache.hadoop.fs.LocalFileSystem";
 
-public class LocalFileSystem implements FileSystem {
-    @Override
-    public void deleteFile(String path) throws IOException {
-        File file = new File(path);
-        file.delete();
-    }
-
-    @Override
-    public List<String> dirList(String dirPath) throws IOException {
-        File file = new File(dirPath);
-        String[] list = file.list();
-        if (list == null) {
-            return null;
-        }
-        return Arrays.asList(list);
+    public LocalConf(String hdfsNameKey) {
+        super(hdfsNameKey);
     }
 }
diff --git a/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/LocalFileSink.java b/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/LocalFileSink.java
@@ -17,16 +17,28 @@
 
 package org.apache.seatunnel.connectors.seatunnel.file.local.sink;
 
+import org.apache.seatunnel.api.common.PrepareFailException;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
-import org.apache.seatunnel.connectors.seatunnel.file.sink.AbstractFileSink;
-import org.apache.seatunnel.connectors.seatunnel.file.sink.spi.SinkFileSystemPlugin;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType;
+import org.apache.seatunnel.connectors.seatunnel.file.local.config.LocalConf;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.BaseFileSink;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import com.google.auto.service.AutoService;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 
 @AutoService(SeaTunnelSink.class)
-public class LocalFileSink extends AbstractFileSink {
+public class LocalFileSink extends BaseFileSink {
+
+    @Override
+    public String getPluginName() {
+        return FileSystemType.LOCAL.getFileSystemPluginName();
+    }
+
     @Override
-    public SinkFileSystemPlugin getSinkFileSystemPlugin() {
-        return new LocalFileSinkPlugin();
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        super.prepare(pluginConfig);
+        hadoopConf = new LocalConf(CommonConfigurationKeys.FS_DEFAULT_NAME_DEFAULT);
     }
 }
diff --git a/...n/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/LocalFileSinkPlugin.java b/...n/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/LocalFileSinkPlugin.java
diff --git a/...e/seatunnel/connectors/seatunnel/file/local/sink/filesystem/LocalFileSystemCommitter.java b/...e/seatunnel/connectors/seatunnel/file/local/sink/filesystem/LocalFileSystemCommitter.java
diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/util/FileUtils.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/sink/util/FileUtils.java