Use apache http client and implement connection pooling

Author: testower

gbfs-validator-java-api/src/main/java/org/entur/gbfs/validator/api/handler/ValidateApiDelegateHandler.java

@@ -22,6 +22,7 @@
 
 import com.google.common.collect.Multimap;
 import com.google.common.collect.MultimapBuilder;
+import jakarta.annotation.PreDestroy;
 import org.entur.gbfs.validation.GbfsValidator;
 import org.entur.gbfs.validation.GbfsValidatorFactory;
 import org.entur.gbfs.validation.model.FileValidationError;
@@ -36,6 +37,8 @@
 import org.entur.gbfs.validator.loader.LoadedFile;
 import org.entur.gbfs.validator.loader.Loader;
 import org.openapitools.jackson.nullable.JsonNullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 
@@ -48,13 +51,36 @@
 
 @Service
 public class ValidateApiDelegateHandler implements ValidateApiDelegate {
+    private static final Logger logger = LoggerFactory.getLogger(ValidateApiDelegateHandler.class);
+
+    // Single shared instance of Loader
+    private final Loader loader;
+
+    // Initialize in constructor
+    public ValidateApiDelegateHandler() {
+        this.loader = new Loader();
+    }
+
+    // Clean up resources when the service is destroyed
+    @PreDestroy
+    public void destroy() {
+        try {
+            if (loader != null) {
+                loader.close();
+            }
+        } catch (IOException e) {
+            logger.error("Error closing Loader", e);
+        }
+    }
 
     @Override
     public ResponseEntity<org.entur.gbfs.validator.api.model.ValidationResult> validatePost(ValidatePostRequest validatePostRequest) {
-        Loader loader = new Loader();
+        logger.debug("Received request for url: {}", validatePostRequest.getFeedUrl());
         try {
             List<LoadedFile> loadedFiles = loader.load(validatePostRequest.getFeedUrl());
 
+            logger.debug("Loaded files: {}", loadedFiles.size());
+
             Multimap<String, LoadedFile> fileMap = MultimapBuilder.hashKeys().arrayListValues().build();
             for (LoadedFile loadedFile : loadedFiles) {
                 fileMap.put(loadedFile.language(), loadedFile);
@@ -66,12 +92,14 @@ public ResponseEntity<org.entur.gbfs.validator.api.model.ValidationResult> valid
             // and then merge the results
             List<org.entur.gbfs.validator.api.model.ValidationResult> results = new ArrayList<>();
             fileMap.keySet().forEach(language -> {
+                logger.debug("Validating language: {}", language);
                 Map<String, InputStream> validatorInputMap = new HashMap<>();
                 Map<String, String> urlMap = new HashMap<>();
                 fileMap.get(language).forEach(file -> {
                     validatorInputMap.put(file.fileName(), file.fileContents());
                     urlMap.put(file.fileName(), file.url());
                 });
+
                 results.add(
                         mapValidationResult(
                                 validator.validate(
@@ -81,6 +109,8 @@ public ResponseEntity<org.entur.gbfs.validator.api.model.ValidationResult> valid
                                 language
                         )
                 );
+
+                logger.debug("Validated {} files for language: {}", validatorInputMap.size(), language);
             });
 
             // merge the list of ValidationResult into a single validation result

gbfs-validator-java-api/src/main/resources/application.properties

@@ -1 +1,5 @@
 spring.application.name=gbfs-validator-java-api
+
+logging.level.org.entur.gbfs.validator.api.handler.ValidateApiDelegateHandler=DEBUG
+server.compression.enabled=true
+server.compression.min-response-size=1024

gbfs-validator-java-loader/pom.xml

@@ -28,6 +28,11 @@
             <artifactId>commons-io</artifactId>
             <version>2.17.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents.client5</groupId>
+            <artifactId>httpclient5</artifactId>
+            <version>5.3.1</version>
+        </dependency>
     </dependencies>
 
 </project>

gbfs-validator-java-loader/src/main/java/org/entur/gbfs/validator/loader/Loader.java

@@ -20,30 +20,70 @@
 
 package org.entur.gbfs.validator.loader;
 
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.hc.core5.http.ParseException;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.util.Timeout;
 import org.json.JSONObject;
 import org.json.JSONTokener;
 
-import java.io.BufferedReader;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.net.HttpURLConnection;
 import java.net.URI;
-import java.net.URL;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 public class Loader {
+    private final CloseableHttpClient httpClient;
+    private final ExecutorService executorService;
+
+    public Loader() {
+        // Create connection pool manager
+        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
+        // Set the maximum number of total connections
+        // TODO configurable max total
+        connectionManager.setMaxTotal(50);
+        // Set the maximum number of connections per route
+        // TODO configurable max per route
+        connectionManager.setDefaultMaxPerRoute(20);
+
+        // Configure request timeouts
+        RequestConfig requestConfig = RequestConfig.custom()
+                // TODO configurable timeouts
+                .setConnectTimeout(Timeout.of(5, TimeUnit.SECONDS))
+                .setResponseTimeout(Timeout.of(5, TimeUnit.SECONDS))
+                .build();
+
+        // Build the HttpClient with connection pooling
+        httpClient = HttpClients.custom()
+                .setConnectionManager(connectionManager)
+                .setDefaultRequestConfig(requestConfig)
+                .build();
+
+        // Create a thread pool for parallel execution
+        // TODO configurable pool size
+        executorService = Executors.newFixedThreadPool(20);
+    }
 
     public List<LoadedFile> load(String discoveryURI) throws IOException {
         InputStream discoveryFileStream = loadFile(URI.create(discoveryURI));
 
+
         ByteArrayOutputStream discoveryFileCopy = new ByteArrayOutputStream();
         org.apache.commons.io.IOUtils.copy(discoveryFileStream, discoveryFileCopy);
         byte[] discoveryFileBytes = discoveryFileCopy.toByteArray();
@@ -74,23 +114,33 @@ private List<LoadedFile> getV3Files(JSONObject discoveryFile, String discoveryFi
                         new ByteArrayInputStream(discoveryFileBytes)
                 ));
 
-        loadedFiles.addAll(
-                discoveryFile.getJSONObject("data").getJSONArray("feeds").toList().stream().map(feed -> {
-            var feedObj = (HashMap) feed;
-            var url = (String) feedObj.get("url");
-            var file = loadFile(URI.create(url));
-            return new LoadedFile(
-                    (String) feedObj.get("name"),
-                    url,
-                    file
-            );
-        }).toList());
+        // Load files in parallel using CompletableFuture
+        List<CompletableFuture<LoadedFile>> futures = discoveryFile.getJSONObject("data").getJSONArray("feeds").toList().stream()
+                .map(feed -> {
+                    var feedObj = (HashMap) feed;
+                    var url = (String) feedObj.get("url");
+                    var name = (String) feedObj.get("name");
+
+                    // Create a CompletableFuture for each file to load
+                    return CompletableFuture.supplyAsync(() -> {
+                        var file = loadFile(URI.create(url));
+                        return new LoadedFile(name, url, file);
+                    }, executorService);
+                })
+                .toList();
+
+        // Wait for all futures to complete and collect results
+        loadedFiles.addAll(futures.stream()
+                .map(CompletableFuture::join)
+                .toList());
 
         return loadedFiles;
     }
 
     private List<LoadedFile> getPreV3Files(JSONObject discoveryFile, String discoveryFileUrl, byte[] discoveryFileBytes) {
         List<LoadedFile> result = new ArrayList<>();
+        List<CompletableFuture<LoadedFile>> futures = new ArrayList<>();
+
         discoveryFile.getJSONObject("data")
                 .keys()
                 .forEachRemaining(key -> {
@@ -102,21 +152,26 @@ private List<LoadedFile> getPreV3Files(JSONObject discoveryFile, String discover
                                     key
                             )
                     );
+
+                    // Create CompletableFutures for each feed file
                     discoveryFile.getJSONObject("data").getJSONObject(key).getJSONArray("feeds").toList().forEach(feed -> {
                         var feedObj = (HashMap) feed;
                         var url = (String) feedObj.get("url");
-                        var file = loadFile(URI.create(url));
-                        result.add(
-                                new LoadedFile(
-                                        (String) feedObj.get("name"),
-                                        url,
-                                        file,
-                                        key
-                                )
-                        );
+                        var name = (String) feedObj.get("name");
+
+                        // Create a CompletableFuture for each file to load
+                        futures.add(CompletableFuture.supplyAsync(() -> {
+                            var file = loadFile(URI.create(url));
+                            return new LoadedFile(name, url, file, key);
+                        }, executorService));
                     });
                 });
 
+        // Wait for all futures to complete and collect results
+        result.addAll(futures.stream()
+                .map(CompletableFuture::join)
+                .toList());
+
         return result;
     }
 
@@ -143,23 +198,39 @@ private static FileInputStream getFileInputStream(URI fileURI) throws FileNotFou
     }
 
     private InputStream getHTTPInputStream(URI fileURI) throws IOException {
-        URL url = fileURI.toURL();
-        HttpURLConnection con = (HttpURLConnection) url.openConnection();
-        con.setRequestMethod("GET");
-        con.setConnectTimeout(5000);
-        con.setReadTimeout(5000);
-        con.setRequestProperty("Et-Client-Name", "entur-gbfs-validator");
-        con.connect();
-        BufferedReader in = new BufferedReader(
-                new InputStreamReader(con.getInputStream()));
-
-        String inputLine;
-        StringBuffer content = new StringBuffer();
-        while ((inputLine = in.readLine()) != null) {
-            content.append(inputLine);
+        HttpGet httpGet = new HttpGet(fileURI);
+
+        // TODO configurable headers
+        httpGet.setHeader("Et-Client-Name", "entur-gbfs-validator");
+
+        try (CloseableHttpResponse response = httpClient.execute(httpGet)) {
+            String content = EntityUtils.toString(response.getEntity());
+            return new ByteArrayInputStream(content.getBytes());
+        } catch (ParseException e) {
+
+            // Todo handle parse exception
+            throw new RuntimeException(e);
+        }
+    }
+
+    // Close the connection pool when the application shuts down
+    public void close() throws IOException {
+        if (httpClient != null) {
+            httpClient.close();
+        }
+
+        // Shutdown the executor service
+        if (executorService != null && !executorService.isShutdown()) {
+            executorService.shutdown();
+            try {
+                // Wait for tasks to complete
+                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+                    executorService.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executorService.shutdownNow();
+                Thread.currentThread().interrupt();
+            }
         }
-        in.close();
-        con.disconnect();
-        return new ByteArrayInputStream(content.toString().getBytes());
     }
 }