Merge pull request #18 from LakshanSS/siddhi-4.x.x

Add support for sqlserver and postgresql in listening mode
siddhi-io · May 15, 2019 · 98b018d · 98b018d
2 parents e2ae032 + 54d6aad
commit 98b018d
Show file tree

Hide file tree

Showing 5 changed files with 79 additions and 7 deletions.
diff --git a/component/pom.xml b/component/pom.xml
@@ -39,10 +39,18 @@
             <groupId>io.debezium</groupId>
             <artifactId>debezium-connector-mysql</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-sqlserver</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.debezium</groupId>
             <artifactId>debezium-connector-oracle</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-postgres</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.wso2.siddhi</groupId>
             <artifactId>siddhi-core</artifactId>

diff --git a/component/src/main/java/org/wso2/extension/siddhi/io/cdc/source/CDCSource.java b/component/src/main/java/org/wso2/extension/siddhi/io/cdc/source/CDCSource.java
@@ -252,8 +252,7 @@
                                 "\ndefine stream inputStream (name string);",
                         description = "In this example, the CDC source polls the 'students' table for inserts " +
                                 "and updates. The polling column is a timestamp field."
-                ),
-
+                )
         }
 )
 

diff --git a/component/src/main/java/org/wso2/extension/siddhi/io/cdc/util/CDCSourceConstants.java b/component/src/main/java/org/wso2/extension/siddhi/io/cdc/util/CDCSourceConstants.java
@@ -41,13 +41,16 @@ public class CDCSourceConstants {
     public static final String CONNECTOR_CLASS = "connector.class";
     public static final String DATABASE_PORT = "database.port";
     public static final String TABLE_WHITELIST = "table.whitelist";
+    public static final String DATABASE_DBNAME = "database.dbname";
     public static final String DATABASE_HOSTNAME = "database.hostname";
     public static final String DATABASE_USER = "database.user";
     public static final String DATABASE_PASSWORD = "database.password";
     public static final String OFFSET_STORAGE = "offset.storage";
     public static final String CDC_SOURCE_OBJECT = "cdc.source.object";
     public static final String DATABASE_HISTORY = "database.history";
     public static final String MYSQL_CONNECTOR_CLASS = "io.debezium.connector.mysql.MySqlConnector";
+    public static final String POSTGRESQL_CONNECTOR_CLASS = "io.debezium.connector.postgresql.PostgresConnector";
+    public static final String SQLSERVER_CONNECTOR_CLASS = "io.debezium.connector.sqlserver.SqlServerConnector";
     public static final String BEFORE_PREFIX = "before_";
     public static final String CACHE_OBJECT = "cacheObj";
     public static final int DEFAULT_SERVER_ID = -1;

diff --git a/component/src/main/java/org/wso2/extension/siddhi/io/cdc/util/CDCSourceUtil.java b/component/src/main/java/org/wso2/extension/siddhi/io/cdc/util/CDCSourceUtil.java
@@ -73,12 +73,62 @@ public static Map<String, Object> getConfigMap(String username, String password,
 
                     //Add other MySQL specific details to configMap.
                     configMap.put(CDCSourceConstants.CONNECTOR_CLASS, CDCSourceConstants.MYSQL_CONNECTOR_CLASS);
+                    break;
+                }
+                case "postgresql": {
+                    //Extract url details
+                    String regex = "jdbc:postgresql://(\\w*|[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}):" +
+                            "(\\d++)/(\\w*)";
+                    Pattern p = Pattern.compile(regex);
+                    Matcher matcher = p.matcher(url);
+                    if (matcher.find()) {
+                        host = matcher.group(1);
+                        port = Integer.parseInt(matcher.group(2));
+                        database = matcher.group(3);
+                    } else {
+                        throw new WrongConfigurationException("Invalid JDBC url: " + url + " received for stream: " +
+                                siddhiStreamName + ". Expected url format: jdbc:postgresql://<host>:<port>/" +
+                                "<database_name>");
+                    }
+
+                    //Add extracted url details to configMap.
+                    configMap.put(CDCSourceConstants.DATABASE_HOSTNAME, host);
+                    configMap.put(CDCSourceConstants.DATABASE_PORT, port);
+                    configMap.put(CDCSourceConstants.DATABASE_DBNAME, database);
+                    configMap.put(CDCSourceConstants.TABLE_WHITELIST, tableName);
+
+                    //Add other PostgreSQL specific details to configMap.
+                    configMap.put(CDCSourceConstants.CONNECTOR_CLASS, CDCSourceConstants.POSTGRESQL_CONNECTOR_CLASS);
+                    break;
+                }
+                case "sqlserver": {
+                    //Extract url details
+                    String regex = "jdbc:sqlserver://(\\w*|[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}):" +
+                            "(\\d++);databaseName=(\\w*)";
+                    Pattern p = Pattern.compile(regex);
+                    Matcher matcher = p.matcher(url);
+                    if (matcher.find()) {
+                        host = matcher.group(1);
+                        port = Integer.parseInt(matcher.group(2));
+                        database = matcher.group(3);
+                    } else {
+                        throw new WrongConfigurationException("Invalid JDBC url: " + url + " received for stream: " +
+                                siddhiStreamName + ". Expected url format: jdbc:sqlserver://<host>:<port>;" +
+                                "databaseName=<database_name>");
+                    }
+                    //Add extracted url details to configMap.
+                    configMap.put(CDCSourceConstants.DATABASE_HOSTNAME, host);
+                    configMap.put(CDCSourceConstants.DATABASE_PORT, port);
+                    configMap.put(CDCSourceConstants.TABLE_WHITELIST, tableName);
+                    configMap.put(CDCSourceConstants.DATABASE_DBNAME, database);
 
+                    //Add other SQLServer specific details to configMap.
+                    configMap.put(CDCSourceConstants.CONNECTOR_CLASS, CDCSourceConstants.SQLSERVER_CONNECTOR_CLASS);
                     break;
                 }
                 default: {
-                    throw new WrongConfigurationException("Unsupported schema. Expected schema: mysql, Found: "
-                            + splittedURL[1]);
+                    throw new WrongConfigurationException("Unsupported schema. Expected schema: mysql or postgresql" +
+                            "or sqlserver, Found: " + splittedURL[1]);
                 }
             }
 

diff --git a/pom.xml b/pom.xml
@@ -68,6 +68,16 @@
                 <artifactId>debezium-connector-oracle</artifactId>
                 <version>${debezium-connector-oracle.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-connector-postgres</artifactId>
+                <version>${debezium-connector-postgres.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-connector-sqlserver</artifactId>
+                <version>${debezium-connector-sqlserver.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.zaxxer</groupId>
                 <artifactId>HikariCP</artifactId>
@@ -191,9 +201,11 @@
         <siddhi-map-keyvalue.version>1.0.13</siddhi-map-keyvalue.version>
         <jacoco.plugin.version>0.7.9</jacoco.plugin.version>
         <mysql-binlog-connector-java.version>0.13.0</mysql-binlog-connector-java.version>
-        <debezium-connector-mysql.version>0.8.3.Final</debezium-connector-mysql.version>
-        <debezium-embedded.version>0.8.3.Final</debezium-embedded.version>
-        <debezium-connector-oracle.version>0.8.3.Final</debezium-connector-oracle.version>
+        <debezium-connector-mysql.version>0.9.5.Final</debezium-connector-mysql.version>
+        <debezium-embedded.version>0.9.5.Final</debezium-embedded.version>
+        <debezium-connector-oracle.version>0.9.5.Final</debezium-connector-oracle.version>
+        <debezium-connector-postgres.version>0.9.5.Final</debezium-connector-postgres.version>
+        <debezium-connector-sqlserver.version>0.9.5.Final</debezium-connector-sqlserver.version>
         <hikari.version>3.2.0</hikari.version>
         <org.testng.version>6.11</org.testng.version>
         <siddhi-store-rdbms.version>4.0.39</siddhi-store-rdbms.version>