Bulk lookup implementation, part 3

- Optimize DNS based bulk lookups

src/main/groovy/io/kamax/mxisd/lookup/provider/DnsLookupProvider.groovy

@@ -33,6 +33,8 @@ import org.xbill.DNS.Lookup
 import org.xbill.DNS.SRVRecord
 import org.xbill.DNS.Type
 
+import java.util.concurrent.ForkJoinPool
+import java.util.concurrent.RecursiveTask
 import java.util.function.Function
 
 @Component
@@ -129,7 +131,6 @@ class DnsLookupProvider extends RemoteIdentityServerProvider {
 
     @Override
     List<ThreePidMapping> populate(List<ThreePidMapping> mappings) {
-        List<ThreePidMapping> mappingsFound = new ArrayList<>()
         Map<String, List<ThreePidMapping>> domains = new HashMap<>()
 
         for (ThreePidMapping mapping : mappings) {
@@ -157,20 +158,59 @@ class DnsLookupProvider extends RemoteIdentityServerProvider {
         }
 
         log.info("Looking mappings across {} domains", domains.keySet().size())
+        ForkJoinPool pool = new ForkJoinPool()
+        RecursiveTask<List<ThreePidMapping>> task = new RecursiveTask<List<ThreePidMapping>>() {
+
+            @Override
+            protected List<ThreePidMapping> compute() {
+                List<ThreePidMapping> mappingsFound = new ArrayList<>()
+                List<DomainBulkLookupTask> tasks = new ArrayList<>()
+
                 for (String domain : domains.keySet()) {
-            Optional<String> baseUrl = findIdentityServerForDomain(domain)
-            if (!baseUrl.isPresent()) {
-                log.info("No usable Identity server for domain {}", domain)
-                continue
+                    DomainBulkLookupTask domainTask = new DomainBulkLookupTask(domain, domains.get(domain))
+                    domainTask.fork()
+                    tasks.add(domainTask)
                 }
 
-            List<ThreePidMapping> domainMappings = find(baseUrl.get(), domains.get(domain))
-            log.info("Found {} mappings in domain {}", domainMappings.size(), domain)
-            mappingsFound.addAll(domainMappings)
+                for (DomainBulkLookupTask task : tasks) {
+                    mappingsFound.addAll(task.join())
                 }
 
+                return mappingsFound
+            }
+        }
+        pool.submit(task)
+        pool.shutdown()
+
+        List<ThreePidMapping> mappingsFound = task.join()
         log.info("Found {} mappings overall", mappingsFound.size())
         return mappingsFound
     }
 
+    private class DomainBulkLookupTask extends RecursiveTask<List<ThreePidMapping>> {
+
+        private String domain
+        private List<ThreePidMapping> mappings
+
+        DomainBulkLookupTask(String domain, List<ThreePidMapping> mappings) {
+            this.domain = domain
+            this.mappings = mappings
+        }
+
+        @Override
+        protected List<ThreePidMapping> compute() {
+            List<ThreePidMapping> domainMappings = new ArrayList<>()
+
+            Optional<String> baseUrl = findIdentityServerForDomain(domain)
+            if (!baseUrl.isPresent()) {
+                log.info("No usable Identity server for domain {}", domain)
+            } else {
+                domainMappings.addAll(find(baseUrl.get(), mappings))
+                log.info("Found {} mappings in domain {}", domainMappings.size(), domain)
+            }
+
+            return domainMappings
+        }
+    }
+
 }

src/main/groovy/io/kamax/mxisd/lookup/provider/RemoteIdentityServerProvider.groovy

@@ -55,6 +55,8 @@ abstract class RemoteIdentityServerProvider implements ThreePidProvider {
             HttpURLConnection rootSrvConn = (HttpURLConnection) new URL(
                     "${remote}/_matrix/identity/api/v1/lookup?medium=${THREEPID_TEST_MEDIUM}&address=${THREEPID_TEST_ADDRESS}"
             ).openConnection()
+            // TODO turn this into a configuration property
+            rootSrvConn.setConnectTimeout(2000)
 
             if (rootSrvConn.getResponseCode() != 200) {
                 return false