[ISSUE #4635] Implemented the functions of file source connector

eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java

@@ -21,8 +21,10 @@
 
 import lombok.Data;
 import lombok.EqualsAndHashCode;
+import lombok.extern.slf4j.Slf4j;
 
 @Data
+@Slf4j
 @EqualsAndHashCode(callSuper = true)
 public class FileSourceConfig extends SourceConfig {
 

eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/SourceConnectorConfig.java

@@ -29,4 +29,9 @@ public class SourceConnectorConfig {
     private String topic;
 
     private long commitOffsetIntervalMs = 5 * 1000;
+
+    private String fileName;
+
+    private String filePath;
+
 }

eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

@@ -23,9 +23,22 @@
 import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
 import org.apache.eventmesh.openconnect.api.source.Source;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReader;
 
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Calendar;
 import java.util.List;
+import java.util.Locale;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -36,6 +49,10 @@ public class FileSourceConnector implements Source {
 
     private OffsetStorageReader offsetStorageReader;
 
+    private String filePath;
+    private String fileName;
+    private BufferedReader bufferedReader;
+
     @Override
     public Class<? extends Config> configClass() {
         return FileSourceConfig.class;
@@ -45,25 +62,28 @@ public Class<? extends Config> configClass() {
     public void init(Config config) throws Exception {
         // init config for hdfs source connector
         this.sourceConfig = (FileSourceConfig) config;
-
+        this.filePath = buildFilePath(((FileSourceConfig) config).getConnectorConfig().getFilePath());
+        this.fileName = buildFileName(((FileSourceConfig) config).getConnectorConfig().getFileName());
     }
 
     @Override
     public void init(ConnectorContext connectorContext) throws Exception {
         SourceConnectorContext sourceConnectorContext = (SourceConnectorContext) connectorContext;
         this.sourceConfig = (FileSourceConfig) sourceConnectorContext.getSourceConfig();
         this.offsetStorageReader = sourceConnectorContext.getOffsetStorageReader();
-
     }
 
     @Override
     public void start() throws Exception {
-
+        if (fileName == null || fileName.isEmpty() || filePath == null || filePath.isEmpty()) {
+            this.bufferedReader = new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8));
+        } else {
+            this.bufferedReader = Files.newBufferedReader(Paths.get(filePath, fileName), StandardCharsets.UTF_8);
+        }
     }
 
     @Override
     public void commit(ConnectRecord record) {
-
     }
 
     @Override
@@ -73,12 +93,48 @@ public String name() {
 
     @Override
     public void stop() {
-
+        try {
+            if (bufferedReader != null) {
+                bufferedReader.close();
+            }
+        } catch (Exception e) {
+            log.error("Error closing resources: {}", e.getMessage());
+        }
     }
 
     @Override
     public List<ConnectRecord> poll() {
-        return null;
+        List<ConnectRecord> connectRecords = new ArrayList<>();
+        try {
+            String line;
+            while ((line = bufferedReader.readLine()) != null) {
+                ConnectRecord connectRecord = new ConnectRecord(new RecordPartition(), new RecordOffset(), System.currentTimeMillis(), line);
+                connectRecords.add(connectRecord);
+            }
+        } catch (IOException e) {
+            log.error("Error reading data from the file: {}", e.getMessage());
+        }
+        return connectRecords;
+    }
+
+    private String buildFilePath(String filePath) {
+        Calendar calendar = Calendar.getInstance(Locale.CHINA);
+        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd");
+        String formattedDate = dateFormat.format(calendar.getTime());
+        String filePath1 = filePath.replace("%s", formattedDate);
+        File path = new File(filePath1);
+        if (!path.exists()) {
+            if (!path.mkdirs()) {
+                log.error("make file dir {} error", filePath);
+            }
+        }
+        return filePath1;
     }
 
+    private String buildFileName(String fileName) {
+        Calendar calendar = Calendar.getInstance(Locale.CHINA);
+        long currentTime = calendar.getTime().getTime();
+        String formattedDate = calendar.get(Calendar.HOUR_OF_DAY) + "-" + currentTime;
+        return fileName.replace("%s", formattedDate);
+    }
 }

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

@@ -15,20 +15,13 @@
 # limitations under the License.
 #
 
-pubSubConfig:
-    meshAddress: 127.0.0.1:10000
-    subject: TopicTest
-    idc: FT
-    env: PRD
-    group: fileSource
-    appId: 5032
-    userName: fileSourceUser
-    passWord: filePassWord
 connectorConfig:
     connectorName: fileSource
     nameserver: 127.0.0.1:9877
     topic: TopicTest
     commitOffsetIntervalMs: 5000
+    fileName: TopicTest-%s
+    filePath: TopicTest/%s
 offsetStorageConfig:
     offsetStorageType: nacos
     offsetStorageAddr: 127.0.0.1:8848