[update][plugin][redisreader] Refactor connection.uri to use a list…

… instead of a string Refactor `connection` to use map instead of a list
wgzhao · Oct 10, 2024 · bcf76c9 · bcf76c9
1 parent ea35dae
commit bcf76c9
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 31 deletions.
diff --git a/docs/assets/jobs/redisreader.json b/docs/assets/jobs/redisreader.json
@@ -5,7 +5,7 @@
         "name": "redisreader",
         "parameter": {
           "connection": {
-            "uri": "tcp://127.0.0.1:7003",
+            "uri": ["tcp://127.0.0.1:6379", "file:///data/dump.rdb", "http://localhost/dump.rdb"],
             "auth": "password"
           },
           "include": [
@@ -23,12 +23,10 @@
       "writer": {
         "name": "rediswriter",
         "parameter": {
-          "connection": [
-            {
-              "uri": "tcp://127.0.0.1:6379",
-              "auth": "123456"
-            }
-          ],
+          "connection": {
+            "uri": "tcp://127.0.0.1:6379",
+            "auth": "123456"
+          },
           "timeout": 60000
         }
       }

diff --git a/...der/redisreader/src/main/java/com/wgzhao/addax/plugin/reader/redisreader/RedisReader.java b/...der/redisreader/src/main/java/com/wgzhao/addax/plugin/reader/redisreader/RedisReader.java
@@ -56,7 +56,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.StringJoiner;
 import java.util.TreeMap;
 import java.util.UUID;
 import java.util.regex.Pattern;
@@ -86,6 +85,7 @@ public static class Job
             extends Reader.Job
     {
         private Configuration conf;
+        private List<String> uris;
 
         @Override
         public void init()
@@ -97,17 +97,19 @@ public void init()
         private void validateParam()
         {
             Configuration conConf = conf.getConfiguration(RedisKey.CONNECTION);
-            String uri = conConf.getString(RedisKey.URI);
-            if (uri == null || uri.isEmpty()) {
-                throw AddaxException.asAddaxException(REQUIRED_VALUE, "uri is null or empty");
-            }
-            if (!(uri.startsWith("tcp") || uri.startsWith("file") || uri.startsWith("http") || uri.startsWith("https"))) {
-                throw AddaxException.asAddaxException(ILLEGAL_VALUE, "uri is not start with tcp, file, http or https");
-            }
-            String mode = conConf.getString(RedisKey.MODE, "standalone");
-            if ("sentinel".equalsIgnoreCase(mode)) {
-                // required other items
-                conConf.getNecessaryValue(RedisKey.MASTER_NAME, REQUIRED_VALUE);
+            uris = conConf.getList(RedisKey.URI, String.class);
+            for (String uri : uris) {
+                if (uri == null || uri.isEmpty()) {
+                    throw AddaxException.asAddaxException(REQUIRED_VALUE, "uri is null or empty");
+                }
+                if (!(uri.startsWith("tcp") || uri.startsWith("file") || uri.startsWith("http") || uri.startsWith("https"))) {
+                    throw AddaxException.asAddaxException(ILLEGAL_VALUE, "uri is not start with tcp, file, http or https");
+                }
+                String mode = conConf.getString(RedisKey.MODE, "standalone");
+                if ("sentinel".equalsIgnoreCase(mode) && uri.startsWith("tcp")) {
+                    // required other items
+                    conConf.getNecessaryValue(RedisKey.MASTER_NAME, REQUIRED_VALUE);
+                }
             }
         }
 
@@ -121,7 +123,23 @@ public void destroy()
         public List<Configuration> split(int adviceNumber)
         {
             // ignore adviceNumber
-            return Collections.singletonList(conf);
+            if (adviceNumber != uris.size() ) {
+                throw AddaxException.asAddaxException(ILLEGAL_VALUE, "adviceNumber is not equal to uri size");
+            }
+            if (adviceNumber == 1) {
+                conf.set(String.format("%s.%s", RedisKey.CONNECTION, RedisKey.URI), uris.get(0));
+                return Collections.singletonList(conf);
+            }
+
+            List<Configuration> configurations = new ArrayList<>();
+            for (String uri : uris) {
+                Configuration clone = conf.clone();
+                Configuration conConf = clone.getConfiguration(RedisKey.CONNECTION);
+                conConf.set(RedisKey.URI, uri);
+                clone.set(RedisKey.CONNECTION, conConf);
+                configurations.add(clone);
+            }
+            return configurations;
         }
     }
 
@@ -176,7 +194,7 @@ else if (uri.startsWith("tcp")) {
                     this.dump(uriToHosts(uri), mode, connection.getString(RedisKey.AUTH), masterName, file);
                 }
                 else {
-                    file = new File(uri);
+                    Files.copy(Paths.get(new URI(uri)), file.toPath());
                 }
 
                 LOG.info("loading {} ", file.getAbsolutePath());
@@ -393,7 +411,7 @@ private List<HostAndPort> uriToHosts(String uris)
                 }
             }
             catch (URISyntaxException e) {
-                e.printStackTrace();
+                throw new RuntimeException(e.getMessage(), e);
             }
             return result;
         }

diff --git a/plugin/reader/redisreader/src/main/resources/plugin_job_template.json b/plugin/reader/redisreader/src/main/resources/plugin_job_template.json
@@ -1,20 +1,13 @@
 {
   "name": "redisreader",
   "parameter": {
-    "connection": [
+    "connection":
       {
-        "uri": "tcp://127.0.0.1:6379",
+        "uri": ["tcp://127.0.0.1:6379", "file:///data/dump.rdb", "http://localhost/dump.rdb"],
         "auth": "",
         "mode": "sentinel|master/slave|standalone|cluster",
         "masterName": "mymaster"
       },
-      {
-        "uri": "file:///data/dump.rdb"
-      },
-      {
-        "uri": "http://localhost/dump.rdb"
-      }
-    ],
     "db": [
       0,
       1