新增PostgreSQL连接器

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/PostgreSQLConfigChecker.java

@@ -0,0 +1,26 @@
+package org.dbsyncer.biz.checker.impl.connector;
+
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.config.PostgreSQLConfig;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+import java.util.Map;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/5 22:14
+ */
+@Component
+public class PostgreSQLConfigChecker extends AbstractDataBaseConfigChecker {
+    @Override
+    public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
+        super.modify(connectorConfig, params);
+        String schema = params.get("schema");
+        Assert.hasText(schema, "Schema is empty.");
+
+        PostgreSQLConfig config = (PostgreSQLConfig) connectorConfig;
+        config.setSchema(schema);
+    }
+}

dbsyncer-connector/pom.xml

@@ -42,6 +42,13 @@
             <artifactId>mssql-jdbc</artifactId>
         </dependency>
 
+        <!-- postgresql -->
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <version>42.3.3</version>
+        </dependency>
+
         <!-- smartcn中文分词器 -->
         <dependency>
             <groupId>org.apache.lucene</groupId>

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/PostgreSQLConfig.java

@@ -0,0 +1,22 @@
+package org.dbsyncer.connector.config;
+
+/**
+ * PostgreSQL连接配置
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/5 23:57
+ */
+public class PostgreSQLConfig extends DatabaseConfig {
+
+    // 构架名
+    private String schema;
+
+    public String getSchema() {
+        return schema;
+    }
+
+    public void setSchema(String schema) {
+        this.schema = schema;
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/SqlServerDatabaseConfig.java

@@ -19,4 +19,4 @@ public class SqlServerDatabaseConfig extends DatabaseConfig {
     public void setSchema(String schema) {
         this.schema = schema;
     }
-}
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/constant/DatabaseConstant.java

@@ -30,4 +30,9 @@ public class DatabaseConstant {
      */
     public static final String SQLSERVER_PAGE_SQL = "SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY %s) AS SQLSERVER_ROW_ID, * FROM (%s) S) A WHERE A.SQLSERVER_ROW_ID BETWEEN ? AND ?";
 
+    //*********************************** PostgreSQL **************************************//
+    /**
+     * PostgreSQL分页语句
+     */
+    public static final String POSTGRESQL_PAGE_SQL = " limit ? OFFSET ?";
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/AbstractDatabaseConnector.java

@@ -26,12 +26,12 @@ import java.sql.*;
 import java.util.*;
 import java.util.stream.Collectors;
 
-public abstract class AbstractDatabaseConnector extends AbstractConnector
+public abstract class AbstractDatabaseConnector<Config> extends AbstractConnector
         implements Connector<DatabaseConnectorMapper, DatabaseConfig>, Database {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    protected abstract String getTableSql(DatabaseConfig config);
+    protected abstract String getTableSql(Config config);
 
     @Override
     public ConnectorMapper connect(DatabaseConfig config) {
@@ -61,7 +61,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
 
     @Override
     public List<Table> getTable(DatabaseConnectorMapper connectorMapper) {
-        String sql = getTableSql(connectorMapper.getConfig());
+        String sql = getTableSql((Config) connectorMapper.getConfig());
         List<String> tableNames = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(sql, String.class));
         if (!CollectionUtils.isEmpty(tableNames)) {
             return tableNames.stream().map(name -> new Table(name)).collect(Collectors.toList());

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/DatabaseConnectorMapper.java

@@ -18,7 +18,7 @@ public class DatabaseConnectorMapper implements ConnectorMapper<DatabaseConfig,
 
     public DatabaseConnectorMapper(DatabaseConfig config) {
         this.config = config;
-        this.dataSource = new SimpleDataSource(config.getUrl(), config.getUsername(), config.getPassword());
+        this.dataSource = new SimpleDataSource(config.getDriverClassName(), config.getUrl(), config.getUsername(), config.getPassword());
     }
 
     public <T> T execute(HandleCallback callback) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/ds/SimpleDataSource.java

@@ -16,11 +16,13 @@ public class SimpleDataSource implements DataSource, AutoCloseable {
 
     private final BlockingQueue<SimpleConnection> pool = new LinkedBlockingQueue<>(2000);
     private long lifeTime = 60 * 1000;
+    private String driverClassName;
     private String url;
     private String username;
     private String password;
 
-    public SimpleDataSource(String url, String username, String password) {
+    public SimpleDataSource(String driverClassName, String url, String username, String password) {
+        this.driverClassName = driverClassName;
         this.url = url;
         this.username = username;
         this.password = password;
@@ -93,7 +95,7 @@ public class SimpleDataSource implements DataSource, AutoCloseable {
     }
 
     private SimpleConnection createConnection() throws SQLException {
-        return new SimpleConnection(this, DatabaseUtil.getConnection(url, username, password));
+        return new SimpleConnection(this, DatabaseUtil.getConnection(driverClassName, url, username, password));
     }
 
     public BlockingQueue<SimpleConnection> getPool() {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/enums/ConnectorEnum.java

@@ -8,6 +8,7 @@ import org.dbsyncer.connector.es.ESConnector;
 import org.dbsyncer.connector.kafka.KafkaConnector;
 import org.dbsyncer.connector.mysql.MysqlConnector;
 import org.dbsyncer.connector.oracle.OracleConnector;
+import org.dbsyncer.connector.postgresql.PostgreSQLConnector;
 import org.dbsyncer.connector.sql.DQLMysqlConnector;
 import org.dbsyncer.connector.sql.DQLOracleConnector;
 import org.dbsyncer.connector.sql.DQLSqlServerConnector;
@@ -34,6 +35,10 @@ public enum ConnectorEnum {
      * SqlServer 连接器
      */
     SQL_SERVER("SqlServer", new SqlServerConnector(), SqlServerDatabaseConfig.class),
+    /**
+     * PostgreSQL 连接器
+     */
+    PostgreSQL("PostgreSQL", new PostgreSQLConnector(), PostgreSQLConfig.class),
     /**
      * Elasticsearch 连接器
      */

dbsyncer-connector/src/main/java/org/dbsyncer/connector/mysql/MysqlConnector.java

@@ -5,7 +5,7 @@ import org.dbsyncer.connector.config.PageSqlConfig;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 
-public final class MysqlConnector extends AbstractDatabaseConnector {
+public final class MysqlConnector extends AbstractDatabaseConnector<DatabaseConfig> {
 
     @Override
     protected String getTableSql(DatabaseConfig config) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/oracle/OracleConnector.java

@@ -13,7 +13,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-public final class OracleConnector extends AbstractDatabaseConnector {
+public final class OracleConnector extends AbstractDatabaseConnector<DatabaseConfig> {
 
     @Override
     protected String getTableSql(DatabaseConfig config) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/postgresql/PostgreSQLConnector.java

@@ -0,0 +1,55 @@
+package org.dbsyncer.connector.postgresql;
+
+import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.connector.config.PageSqlConfig;
+import org.dbsyncer.connector.config.PostgreSQLConfig;
+import org.dbsyncer.connector.config.Table;
+import org.dbsyncer.connector.constant.DatabaseConstant;
+import org.dbsyncer.connector.database.AbstractDatabaseConnector;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
+import org.dbsyncer.connector.enums.TableTypeEnum;
+
+import java.util.LinkedList;
+import java.util.List;
+
+public final class PostgreSQLConnector extends AbstractDatabaseConnector<PostgreSQLConfig> {
+
+    @Override
+    protected String getTableSql(PostgreSQLConfig config) {
+        return String.format("SELECT TABLENAME FROM PG_TABLES WHERE SCHEMANAME ='%s'", config.getSchema());
+    }
+
+    @Override
+    public List<Table> getTable(DatabaseConnectorMapper connectorMapper) {
+        List<Table> list = new LinkedList<>();
+        PostgreSQLConfig config = (PostgreSQLConfig) connectorMapper.getConfig();
+        List<String> tableNames = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(getTableSql(config), String.class));
+        if (!CollectionUtils.isEmpty(tableNames)) {
+            tableNames.forEach(name -> list.add(new Table(name, TableTypeEnum.TABLE.getCode())));
+        }
+        List<String> tableViewNames = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(getTableViewSql(config), String.class));
+        if (!CollectionUtils.isEmpty(tableViewNames)) {
+            tableViewNames.forEach(name -> list.add(new Table(name, TableTypeEnum.VIEW.getCode())));
+        }
+        return list;
+    }
+
+    @Override
+    public String getPageSql(PageSqlConfig config) {
+        return config.getQuerySql() + DatabaseConstant.POSTGRESQL_PAGE_SQL;
+    }
+
+    @Override
+    public Object[] getPageArgs(int pageIndex, int pageSize) {
+        return new Object[]{pageSize, (pageIndex - 1) * pageSize};
+    }
+
+    @Override
+    protected String buildSqlWithQuotation() {
+        return "\"";
+    }
+
+    private String getTableViewSql(PostgreSQLConfig config) {
+        return String.format("SELECT VIEWNAME FROM PG_VIEWS WHERE SCHEMANAME ='%s'", config.getSchema());
+    }
+}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLMysqlConnector.java

@@ -8,7 +8,7 @@ import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import java.util.List;
 import java.util.Map;
 
-public final class DQLMysqlConnector extends AbstractDatabaseConnector {
+public final class DQLMysqlConnector extends AbstractDatabaseConnector<DatabaseConfig> {
 
     @Override
     protected String getTableSql(DatabaseConfig config) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLOracleConnector.java

@@ -8,7 +8,7 @@ import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import java.util.List;
 import java.util.Map;
 
-public final class DQLOracleConnector extends AbstractDatabaseConnector {
+public final class DQLOracleConnector extends AbstractDatabaseConnector<DatabaseConfig> {
 
     @Override
     protected String getTableSql(DatabaseConfig config) {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLSqlServerConnector.java

@@ -14,7 +14,7 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 import java.util.Map;
 
-public final class DQLSqlServerConnector extends AbstractDatabaseConnector {
+public final class DQLSqlServerConnector extends AbstractDatabaseConnector<SqlServerDatabaseConfig> {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -29,9 +29,8 @@ public final class DQLSqlServerConnector extends AbstractDatabaseConnector {
     }
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
-        SqlServerDatabaseConfig cfg = (SqlServerDatabaseConfig) config;
-        return String.format("SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('%s')", cfg.getSchema());
+    protected String getTableSql(SqlServerDatabaseConfig config) {
+        return String.format("SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('%s')", config.getSchema());
     }
 
     @Override

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServer.java

@@ -1,5 +0,0 @@
-package org.dbsyncer.connector.sqlserver;
-
-public interface SqlServer {
-
-}

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerConnector.java

@@ -13,7 +13,7 @@ import org.slf4j.LoggerFactory;
 import java.util.HashMap;
 import java.util.Map;
 
-public final class SqlServerConnector extends AbstractDatabaseConnector implements SqlServer {
+public final class SqlServerConnector extends AbstractDatabaseConnector<SqlServerDatabaseConfig> {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -28,9 +28,8 @@ public final class SqlServerConnector extends AbstractDatabaseConnector implemen
     }
 
     @Override
-    protected String getTableSql(DatabaseConfig config) {
-        SqlServerDatabaseConfig cfg = (SqlServerDatabaseConfig) config;
-        return String.format("SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('%s') AND IS_MS_SHIPPED = 0", cfg.getSchema());
+    protected String getTableSql(SqlServerDatabaseConfig config) {
+        return String.format("SELECT NAME FROM SYS.TABLES WHERE SCHEMA_ID = SCHEMA_ID('%s') AND IS_MS_SHIPPED = 0", config.getSchema());
     }
 
     @Override

dbsyncer-connector/src/main/java/org/dbsyncer/connector/util/DatabaseUtil.java

@@ -1,5 +1,7 @@
 package org.dbsyncer.connector.util;
 
+import org.dbsyncer.connector.ConnectorException;
+
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
@@ -9,7 +11,12 @@ public abstract class DatabaseUtil {
     private DatabaseUtil() {
     }
 
-    public static Connection getConnection(String url, String username, String password) throws SQLException {
+    public static Connection getConnection(String driverClassName, String url, String username, String password) throws SQLException {
+        try {
+            Class.forName(driverClassName);
+        } catch (ClassNotFoundException e) {
+            throw new ConnectorException(e.getCause());
+        }
         return DriverManager.getConnection(url, username, password);
     }
 

dbsyncer-web/src/main/resources/public/connector/addPostgreSQL.html

@@ -0,0 +1,50 @@
+<!DOCTYPE html>
+<html lang="zh-CN"
+      xmlns="http://www.w3.org/1999/xhtml" xmlns:th="http://www.thymeleaf.org">
+
+<div th:fragment="content">
+    <div class="form-group">
+        <label class="col-sm-2 control-label">帐号 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" dbsyncer-valid="require" maxlength="32" name="username" placeholder="root"
+                   th:value="${connector?.config?.username}" type="text"/>
+        </div>
+        <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4 ">
+            <input class="form-control" dbsyncer-valid="require" maxlength="32" name="password"
+                   th:value="${connector?.config?.password}" type="password"/>
+        </div>
+    </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-10">
+            <textarea class="form-control dbsyncer_textarea_resize_none" dbsyncer-valid="require" maxlength="1024"
+                      name="url" rows="5"
+                      th:text="${connector?.config?.url} ?: 'jdbc:postgresql://127.0.0.1:5432/test'"></textarea>
+        </div>
+    </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">架构名 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" dbsyncer-valid="require" maxlength="32" name="schema" placeholder="dbo"
+                   th:value="${connector?.config?.schema} ?: 'public'" type="text"/>
+        </div>
+        <div class="col-sm-6"></div>
+    </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">驱动 </label>
+        <div class="col-sm-10">
+            <input class="form-control" name="driverClassName" readonly="true"
+                   th:value="${connector?.config?.driverClassName} ?: 'org.postgresql.Driver'" type="text"/>
+        </div>
+    </div>
+
+    <script type="text/javascript">
+        $(function () {
+            // 初始化select插件
+            initSelectIndex($(".select-control"), 1);
+        })
+    </script>
+</div>
+
+</html>

pom.xml

@@ -45,6 +45,7 @@
         <mysql.version>5.1.40</mysql.version>
         <mysql-binlog.version>0.21.0</mysql-binlog.version>
         <mssql-jdbc.version>8.2.0.jre8</mssql-jdbc.version>
+        <postgresql.version>42.3.3</postgresql.version>
         <kafka.version>0.9.0.0</kafka.version>
         <json.version>20090211</json.version>
         <fastjson.version>1.2.75</fastjson.version>
@@ -150,6 +151,13 @@
                 <version>${mssql-jdbc.version}</version>
             </dependency>
 
+            <!-- postgresql -->
+            <dependency>
+                <groupId>org.postgresql</groupId>
+                <artifactId>postgresql</artifactId>
+                <version>${postgresql.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>com.github.shyiko</groupId>
                 <artifactId>mysql-binlog-connector-java</artifactId>