Merge pull request #109 from hiroyuki-sato/add_mysql_ssl_option

Add mysql ssl option

Author: hito4t

embulk-input-jdbc/README.md

@@ -14,6 +14,7 @@ Generic JDBC input plugin for Embulk loads records from a database using a JDBC
 - **url**: URL of the JDBC connection (e.g. 'jdbc:sqlite:mydb.sqlite3') (string, required)
 - **user**: database login user name (string, optional)
 - **password**: database login password (string, default: optional)
+- **ssl**: use SSL to connect to the database (string, default: `disable`. `enable` uses SSL without server-side validation and `verify` checks the certificate. For compatibility reasons, `true` behaves as `enable` and `false` behaves as `disable`.)
 - **schema**: destination schema name (string, default: use the default schema)
 - **fetch_rows**: number of rows to fetch one time (integer, default: 10000)
 - **connect_timeout**: not supported.

embulk-input-jdbc/src/main/java/org/embulk/input/jdbc/Ssl.java

@@ -0,0 +1,37 @@
+package org.embulk.input.jdbc;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+import org.embulk.config.ConfigException;
+
+public enum Ssl
+{
+    ENABLE,
+    DISABLE,
+    VERIFY;
+
+    @JsonValue
+    @Override
+    public String toString()
+    {
+        return this.name().toLowerCase();
+    }
+
+    @JsonCreator
+    public static Ssl fromString(String value)
+    {
+        switch(value) {
+        case "enable":
+        case "true":
+            return ENABLE;
+        case "disable":
+        case "false":
+            return DISABLE;
+        case "verify":
+            return VERIFY;
+        default:
+            throw new ConfigException(String.format("Unknown SSL value '%s'. Supported values are enable, true, disable, false or verify.", value));
+        }
+    }
+}

embulk-input-mysql/src/main/java/org/embulk/input/MySQLInputPlugin.java

@@ -12,6 +12,7 @@
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
 import org.embulk.input.jdbc.AbstractJdbcInputPlugin;
+import org.embulk.input.jdbc.Ssl;
 import org.embulk.input.jdbc.JdbcInputConnection;
 import org.embulk.input.jdbc.getter.ColumnGetterFactory;
 import org.embulk.input.mysql.MySQLInputConnection;
@@ -42,6 +43,10 @@ public interface MySQLPluginTask
 
         @Config("database")
         public String getDatabase();
+
+        @Config("ssl")
+        @ConfigDefault("\"disable\"") // backward compatibility
+        public Ssl getSsl();
     }
 
     @Override
@@ -75,21 +80,21 @@ protected MySQLInputConnection newConnection(PluginTask task) throws SQLExceptio
         // Socket options TCP_KEEPCNT, TCP_KEEPIDLE, and TCP_KEEPINTVL are not configurable.
         props.setProperty("tcpKeepAlive", "true");
 
-        // TODO
-        //switch task.getSssl() {
-        //when "disable":
-        //    break;
-        //when "enable":
-        //    props.setProperty("useSSL", "true");
-        //    props.setProperty("requireSSL", "false");
-        //    props.setProperty("verifyServerCertificate", "false");
-        //    break;
-        //when "verify":
-        //    props.setProperty("useSSL", "true");
-        //    props.setProperty("requireSSL", "true");
-        //    props.setProperty("verifyServerCertificate", "true");
-        //    break;
-        //}
+        switch (t.getSsl()) {
+            case DISABLE:
+                props.setProperty("useSSL", "false");
+                break;
+            case ENABLE:
+                props.setProperty("useSSL", "true");
+                props.setProperty("requireSSL", "true");
+                props.setProperty("verifyServerCertificate", "false");
+                break;
+            case VERIFY:
+                props.setProperty("useSSL", "true");
+                props.setProperty("requireSSL", "true");
+                props.setProperty("verifyServerCertificate", "true");
+                break;
+        }
 
         if (t.getFetchRows() == 1) {
             logger.info("Fetch size is 1. Fetching rows one by one.");