Changeset 6131 for SRUAggregator/trunk/src/main/java/eu/clarin/sru/fcs/aggregator/app/Aggregator.java

Timestamp:

04/07/15 20:38:40 (9 years ago)

Author:

emanuel.dima@uni-tuebingen.de

Message:

22. beta-33: support for slow endpoints; POST external searches; UI errors for corner cases

File:

• SRUAggregator/trunk/src/main/java/eu/clarin/sru/fcs/aggregator/app/Aggregator.java (modified) (10 diffs)

SRUAggregator/trunk/src/main/java/eu/clarin/sru/fcs/aggregator/app/Aggregator.java

@@ -8 +8 @@
 import com.optimaize.langdetect.profiles.LanguageProfileReader;
 import com.optimaize.langdetect.text.*;
+import eu.clarin.sru.client.SRUThreadedClient;
 import eu.clarin.sru.fcs.aggregator.search.Search;
 import eu.clarin.sru.fcs.aggregator.scan.ScanCrawlTask;
@@ -14 +15 @@
 import eu.clarin.sru.client.fcs.ClarinFCSClientBuilder;
 import eu.clarin.sru.client.fcs.ClarinFCSEndpointDescriptionParser;
+import eu.clarin.sru.fcs.aggregator.client.MaxConcurrentRequestsCallback;
 import eu.clarin.sru.fcs.aggregator.client.ThrottledClient;
 import eu.clarin.sru.fcs.aggregator.scan.Corpus;
@@ -25 +27 @@
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -65 +68 @@
  * @author edima
  *
- * TODO: improve help page text
- *
  * TODO: update comments everywhere
  *
@@ -102 +103 @@
         private TextObjectFactory textObjectFactory;
 
-        private ThrottledClient sruScanClient = null;
-        private ThrottledClient sruSearchClient = null;
+        private ThrottledClient sruClient = null;
+        public MaxConcurrentRequestsCallback maxScanConcurrentRequestsCallback;
+        public MaxConcurrentRequestsCallback maxSearchConcurrentRequestsCallback;
         private Map<Long, Search> activeSearches = Collections.synchronizedMap(new HashMap<Long, Search>());
 
@@ -173 +175 @@
         public void init() throws IOException {
                 log.info("Aggregator initialization started.");
-                sruScanClient = new ThrottledClient(
-                                new ClarinFCSClientBuilder()
+
+                SRUThreadedClient sruScanClient
+                                = new ClarinFCSClientBuilder()
                                 .setConnectTimeout(params.ENDPOINTS_SCAN_TIMEOUT_MS)
                                 .setSocketTimeout(params.ENDPOINTS_SCAN_TIMEOUT_MS)
@@ -181 +184 @@
                                                 new ClarinFCSEndpointDescriptionParser())
                                 .enableLegacySupport()
-                                .buildThreadedClient(),
-                                params.SCAN_MAX_CONCURRENT_REQUESTS_PER_ENDPOINT
-                );
-                sruSearchClient = new ThrottledClient(
-                                new ClarinFCSClientBuilder()
+                                .buildThreadedClient();
+
+                SRUThreadedClient sruSearchClient
+                                = new ClarinFCSClientBuilder()
                                 .setConnectTimeout(params.ENDPOINTS_SEARCH_TIMEOUT_MS)
                                 .setSocketTimeout(params.ENDPOINTS_SEARCH_TIMEOUT_MS)
                                 .addDefaultDataViewParsers()
+                                .registerExtraResponseDataParser(
+                                                new ClarinFCSEndpointDescriptionParser())
                                 .enableLegacySupport()
-                                .buildThreadedClient(),
-                                params.SEARCH_MAX_CONCURRENT_REQUESTS_PER_ENDPOINT
+                                .buildThreadedClient();
+
+                maxScanConcurrentRequestsCallback = new MaxConcurrentRequestsCallback() {
+                        @Override
+                        public int getMaxConcurrentRequest(URI baseURI) {
+                                return params.SCAN_MAX_CONCURRENT_REQUESTS_PER_ENDPOINT;
+                        }
+                };
+
+                maxSearchConcurrentRequestsCallback = new MaxConcurrentRequestsCallback() {
+                        @Override
+                        public int getMaxConcurrentRequest(URI baseURI) {
+                                return params.slowEndpoints.contains(baseURI)
+                                                ? params.SEARCH_MAX_CONCURRENT_REQUESTS_PER_SLOW_ENDPOINT
+                                                : params.SEARCH_MAX_CONCURRENT_REQUESTS_PER_ENDPOINT;
+                        }
+                };
+
+                sruClient = new ThrottledClient(
+                                sruScanClient, maxScanConcurrentRequestsCallback,
+                                sruSearchClient, maxSearchConcurrentRequestsCallback
                 );
 
@@ -221 +244 @@
                 initLanguageDetector();
 
-                ScanCrawlTask task = new ScanCrawlTask(sruScanClient,
+                ScanCrawlTask task = new ScanCrawlTask(sruClient,
                                 params.CENTER_REGISTRY_URL, params.SCAN_MAX_DEPTH,
                                 params.additionalCQLEndpoints,
@@ -237 +260 @@
                         search.shutdown();
                 }
-                shutdownAndAwaitTermination(config.aggregatorParams, sruScanClient, scheduler);
-                shutdownAndAwaitTermination(config.aggregatorParams, sruSearchClient, scheduler);
+                shutdownAndAwaitTermination(config.aggregatorParams, sruClient, scheduler);
                 log.info("Aggregator shutdown complete.");
         }
@@ -253 +275 @@
                         return null;
                 } else {
-                        Search sr = new Search(sruSearchClient, version, searchStatsAtom.get(),
+                        Search sr = new Search(sruClient,
+                                        version, searchStatsAtom.get(),
                                         corpora, searchString, searchLang, maxRecords);
                         if (activeSearches.size() > SEARCHES_SIZE_GC_THRESHOLD) {