更新

srt-cloud-data-integrate/src/main/java/net/srt/entity/DataDatabaseEntity.java

@@ -1,5 +1,6 @@
 package net.srt.entity;
 
+import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import com.baomidou.mybatisplus.annotation.*;
@@ -78,5 +79,18 @@ public class DataDatabaseEntity extends BaseEntity {
 	 */
 	private Long projectId;
 
+	@Schema(description = "是否是关系型数据库")
+	private String isJdbc;
+
+	@Schema(description = "es索引")
+	private String databaseIndex;
+
+	@Schema(description = "es文档")
+	private String databaseDocument;
+
+
+	@Schema(description = "kafka主题")
+	private String databaseKafKa;
+
 
 }

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/DataDatabaseServiceImpl.java

@@ -54,6 +54,7 @@ import srt.cloud.framework.dbswitch.core.model.TableDescription;
 import srt.cloud.framework.dbswitch.core.service.IMetaDataByJdbcService;
 import srt.cloud.framework.dbswitch.core.service.impl.MetaDataByJdbcServiceImpl;
 
+import javax.annotation.Resource;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -69,11 +70,21 @@ import java.util.stream.Collectors;
  * @since 1.0.0 2022-10-09
  */
 @Service
-@AllArgsConstructor
 public class DataDatabaseServiceImpl extends BaseServiceImpl<DataDatabaseDao, DataDatabaseEntity> implements DataDatabaseService {
-
+	@Resource
-	private final DataAccessDao dataAccessDao;
+	private  DataAccessDao dataAccessDao;
-	private final DataAccessService dataAccessService;
+	@Resource
+	private  DataAccessService dataAccessService;
+	@Resource
+	private  EsService esService;
+	@Resource
+	private  MongoService mongoService;
+	@Resource
+	private  KafkaService kafkaService;
+	@Resource
+	private  RedisService redisService;
+	@Resource
+	private  FtpService ftpService;
 
 	@Override
 	public PageResult<DataDatabaseVO> page(DataDatabaseQuery query) {
@@ -98,6 +109,10 @@ public class DataDatabaseServiceImpl extends BaseServiceImpl<DataDatabaseDao, Da
 	public void save(DataDatabaseVO vo) {
 		DataDatabaseEntity entity = DataDatabaseConvert.INSTANCE.convert(vo);
 		entity.setProjectId(getProjectId());
+		entity.setIsJdbc(vo.getIsJdbc());
+		entity.setDatabaseIndex(vo.getDatabaseIndex());
+		entity.setDatabaseDocument(vo.getDatabaseDocument());
+		entity.setDatabaseKafKa(vo.getDatabaseKafKa());
 		setJdbcUrlByEntity(entity);
 		baseMapper.insert(entity);
 		try {
@@ -326,13 +341,32 @@ public class DataDatabaseServiceImpl extends BaseServiceImpl<DataDatabaseDao, Da
 
 	private List<TableVo> getTables(DataDatabaseEntity dataDatabaseEntity) {
 		ProductTypeEnum productTypeEnum = ProductTypeEnum.getByIndex(dataDatabaseEntity.getDatabaseType());
-		IMetaDataByJdbcService metaDataService = new MetaDataByJdbcServiceImpl(productTypeEnum);
+		List<TableDescription> tableDescriptions = new ArrayList<>();
-		List<TableDescription> tableDescriptions = metaDataService.queryTableList(StringUtil.isBlank(dataDatabaseEntity.getJdbcUrl()) ? productTypeEnum.getUrl()
+		try {
-						.replace("{host}", dataDatabaseEntity.getDatabaseIp())
+			switch (productTypeEnum) {
-						.replace("{port}", dataDatabaseEntity.getDatabasePort())
+				case FTP:
-						.replace("{database}", dataDatabaseEntity.getDatabaseName()) : dataDatabaseEntity.getJdbcUrl(), dataDatabaseEntity.getUserName(), dataDatabaseEntity.getPassword(),
+					tableDescriptions = ftpService.getFtpFiles(dataDatabaseEntity);
-				ProductTypeEnum.ORACLE.equals(productTypeEnum) ? dataDatabaseEntity.getUserName() : dataDatabaseEntity.getDatabaseName());
 					return BeanUtil.copyListProperties(tableDescriptions, TableVo::new);
+				case ES:
+					tableDescriptions = esService.getEsTables(dataDatabaseEntity);
+					return BeanUtil.copyListProperties(tableDescriptions, TableVo::new);
+				case MONGODB:
+					tableDescriptions = mongoService.getMongoTables(dataDatabaseEntity);
+					return BeanUtil.copyListProperties(tableDescriptions, TableVo::new);
+				case REDIS:
+					tableDescriptions = redisService.getRedisTables(dataDatabaseEntity);
+					return BeanUtil.copyListProperties(tableDescriptions, TableVo::new);
+				case KAFKA:
+					tableDescriptions = kafkaService.getKafkaTables(dataDatabaseEntity);
+					return BeanUtil.copyListProperties(tableDescriptions, TableVo::new);
+//				default:
+//					IMetaDataByJdbcService metaDataService = new MetaDataByJdbcServiceImpl(productTypeEnum);
+//					tableDescriptions = metaDataService.queryTableList(StringUtil.isBlank(dataDatabaseEntity.getJdbcUrl()) ? productTypeEnum.getUrl().replace("{host}", dataDatabaseEntity.getDatabaseIp()).replace("{port}", dataDatabaseEntity.getDatabasePort()).replace("{database}", dataDatabaseEntity.getDatabaseName()) : dataDatabaseEntity.getJdbcUrl(), dataDatabaseEntity.getUserName(), dataDatabaseEntity.getPassword(), ProductTypeEnum.ORACLE.equals(productTypeEnum) ? dataDatabaseEntity.getUserName() : dataDatabaseEntity.getDatabaseName());
+			}
+		} catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+		return null;
     }
 
 	@SneakyThrows

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/EsService.java

@@ -0,0 +1,14 @@
+package net.srt.service.impl;
+
+import net.srt.entity.DataDatabaseEntity;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+import srt.cloud.framework.dbswitch.core.model.TableDescription;
+
+import java.util.List;
+@Service
+public class EsService {
+    public List<TableDescription> getEsTables(DataDatabaseEntity dataDatabaseEntity) {
+        return null;
+    }
+}

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/FtpService.java

@@ -0,0 +1,13 @@
+package net.srt.service.impl;
+
+import net.srt.entity.DataDatabaseEntity;
+import org.springframework.stereotype.Service;
+import srt.cloud.framework.dbswitch.core.model.TableDescription;
+
+import java.util.List;
+@Service
+public class FtpService {
+    public List<TableDescription> getFtpFiles(DataDatabaseEntity dataDatabaseEntity) {
+        return null;
+    }
+}

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/KafkaService.java

@@ -0,0 +1,41 @@
+package net.srt.service.impl;
+
+import net.srt.entity.DataDatabaseEntity;
+import net.srt.vo.TableVo;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.common.KafkaFuture;
+import org.springframework.stereotype.Service;
+import srt.cloud.framework.dbswitch.core.model.TableDescription;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+@Service
+public class KafkaService {
+    public List<TableDescription> getKafkaTables(DataDatabaseEntity dataDatabaseEntity) {
+        String bootstrapServers = dataDatabaseEntity.getDatabaseIp() + "." + dataDatabaseEntity.getDatabasePort();
+        Properties properties = new Properties();
+        properties.put("bootstrap.servers",bootstrapServers);
+        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
+        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
+        AdminClient adminClient=null;
+        try {
+            adminClient = KafkaAdminClient.create(properties);
+            ListTopicsResult listTopicsResult = adminClient.listTopics();
+            KafkaFuture<Set<String>> names = listTopicsResult.names();
+            Set<String> strings = names.get();
+            List<TableDescription> tableDescriptions=strings.stream().map(s -> {
+                TableDescription tableDescription = new TableDescription();
+                tableDescription.setTableName(s);
+                return tableDescription;
+            }).collect(Collectors.toList());
+            return tableDescriptions;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+}

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/MongoService.java

@@ -0,0 +1,13 @@
+package net.srt.service.impl;
+
+import net.srt.entity.DataDatabaseEntity;
+import org.springframework.stereotype.Service;
+import srt.cloud.framework.dbswitch.core.model.TableDescription;
+
+import java.util.List;
+@Service
+public class MongoService {
+    public List<TableDescription> getMongoTables(DataDatabaseEntity dataDatabaseEntity) {
+        return null;
+    }
+}

srt-cloud-data-integrate/src/main/java/net/srt/service/impl/RedisService.java

@@ -0,0 +1,13 @@
+package net.srt.service.impl;
+
+import net.srt.entity.DataDatabaseEntity;
+import org.springframework.stereotype.Service;
+import srt.cloud.framework.dbswitch.core.model.TableDescription;
+
+import java.util.List;
+@Service
+public class RedisService {
+    public List<TableDescription> getRedisTables(DataDatabaseEntity dataDatabaseEntity) {
+        return null;
+    }
+}

srt-cloud-data-integrate/src/main/java/net/srt/vo/TableVo.java

@@ -13,4 +13,12 @@ public class TableVo {
 	private String tableName;
 	private String remarks;
 	private DBTableType tableType;
+
+	public static TableVo createKafkaTopicVo(String topicName) {
+		TableVo kafkaTopicVo = new TableVo();
+		kafkaTopicVo.setTableName(topicName);
+		kafkaTopicVo.setRemarks("Kafka Topic");
+		kafkaTopicVo.setTableType(null);
+		return kafkaTopicVo;
+	}
 }