Support for large downloads in Complex Portal

pom.xml

@@ -27,6 +27,7 @@
         <dataexchange.version>4.3.2-SNAPSHOT</dataexchange.version>
         <complex.services.version>1.2.0-SNAPSHOT</complex.services.version>
         <lombok.version>1.18.28</lombok.version>
+        <jackson.version>2.15.2</jackson.version>
     </properties>
 
     <scm>
@@ -210,6 +211,24 @@
             <version>1.9.13</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
         <!-- Castor for XML -->
         <dependency>
             <groupId>org.codehaus.castor</groupId>
@@ -252,6 +271,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>2.23.0</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>
@@ -279,20 +305,15 @@
                 <version>2.1.1</version>
             </plugin>
             <plugin>
-                <groupId>org.mortbay.jetty</groupId>
+                <groupId>org.eclipse.jetty</groupId>
                 <artifactId>jetty-maven-plugin</artifactId>
-                <version>7.2.0.RC0</version>
+                <version>9.4.57.v20241219</version>
                 <configuration>
                     <webAppConfig>
                         <contextPath>/intact/complex-ws</contextPath>
                         <jettyEnvXml>${project.build.directory}/classes/META-INF/jetty-env.xml</jettyEnvXml>
                     </webAppConfig>
                     <scanIntervalSeconds>10</scanIntervalSeconds>
-                    <connectors>
-                        <connector implementation="org.eclipse.jetty.server.nio.SelectChannelConnector">
-                            <port>9090</port>
-                        </connector>
-                    </connectors>
                     <jettyConfig>${project.build.directory}/classes/META-INF/jetty.xml</jettyConfig>
                     <stopPort>9009</stopPort>
                     <stopKey />

src/main/java/uk/ac/ebi/intact/service/complex/ws/ComplexCovariationManager.java

@@ -57,7 +57,7 @@ public class ComplexCovariationManager {
     private final String intactProteinScoresEndpointUrl;
     private final String intactProteinScoreDistributionEndpointUrl;
 
-    public ComplexCovariationManager(IntactDao intactDao, @Value("${intact.graph.ws.url}") String intactGraphWsUrlProperty) throws NamingException {
+    public ComplexCovariationManager(IntactDao intactDao, @Value("${intact.graph.ws.url}") String intactGraphWsUrlProperty) {
         String intactGraphWebServiceUrl = getIntactGraphWsUrl(intactGraphWsUrlProperty);
 
         this.intactDao = intactDao;

src/main/java/uk/ac/ebi/intact/service/complex/ws/ComplexExportManager.java

@@ -0,0 +1,116 @@
+package uk.ac.ebi.intact.service.complex.ws;
+
+import com.google.common.collect.Lists;
+import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.springframework.stereotype.Component;
+import uk.ac.ebi.intact.dataexchange.psimi.solr.complex.ComplexSearchResults;
+import uk.ac.ebi.intact.service.complex.ws.model.ComplexRestResult;
+import uk.ac.ebi.intact.service.complex.ws.writer.ComplexWriter;
+import uk.ac.ebi.intact.service.complex.ws.writer.ComplexWriterFactory;
+import uk.ac.ebi.intact.service.complex.ws.writer.SerialisedComplexesWriter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Log4j
+@Component
+@AllArgsConstructor
+public class ComplexExportManager {
+
+    private static final int COMPLEXES_PARTITION_SIZE = 6_000;
+    private static final int MAX_NUMBER_THREADS = 4;
+
+    private final ComplexManager complexManager;
+    private final ComplexExporter complexExporter;
+    private final ComplexWriterFactory complexWriterFactory;
+
+    public CompletableFuture<String> exportComplexes(String query, String filters, ComplexExportFormat format) {
+        try {
+            List<String> complexesAcs = getComplexesAcs(query, filters);
+            if (!complexesAcs.isEmpty()) {
+                List<List<String>> complexesAcsChunks = Lists.partition(complexesAcs, COMPLEXES_PARTITION_SIZE);
+                int numberOfThreads = Math.max(1, Math.min(MAX_NUMBER_THREADS, complexesAcsChunks.size()));
+                ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
+
+                List<CompletableFuture<String>> futures = startSerialiseComplexesThreads(executor, complexesAcsChunks, format);
+                CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+
+                // The writer needs to be created before the async logic to use the right XML writer factory
+                SerialisedComplexesWriter mainComplexWriter = complexWriterFactory.getSerialisedComplexesWriter(format);
+                return allFutures.thenApplyAsync(v -> {
+                    try {
+                        return combineAllSerialisedContent(mainComplexWriter, futures);
+                    } catch (IOException | ExecutionException | InterruptedException e) {
+                        throw new RuntimeException("Export failed " + query + ". Failed to serialise complexes");
+                    } catch (Throwable e) {
+                        log.error(e.getMessage(), e);
+                        throw new RuntimeException(e);
+                    } finally {
+                        executor.shutdown();
+                    }
+                });
+            } else {
+                throw new RuntimeException("Export failed " + query + ". No complexes result");
+            }
+        } catch (SolrServerException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private List<String> getComplexesAcs(String query, String filters) throws SolrServerException {
+        List<String> complexesAcs;
+        if (isQueryASingleId(query)) {
+            complexesAcs = new ArrayList<>(1);
+            complexesAcs.add(query);
+        } else {
+            ComplexRestResult searchResult = complexManager.query(query, null, null, filters, null);
+
+            complexesAcs = new ArrayList<>(searchResult.getElements().size());
+            for (ComplexSearchResults result : searchResult.getElements()) {
+                complexesAcs.add(result.getComplexAC());
+            }
+        }
+        return complexesAcs;
+    }
+
+    private List<CompletableFuture<String>> startSerialiseComplexesThreads(
+            ExecutorService executor,
+            List<List<String>> complexesAcsChunks,
+            ComplexExportFormat format) {
+
+        List<CompletableFuture<String>> futures = new ArrayList<>();
+        for (List<String> complexesAcsChunk : complexesAcsChunks) {
+            // The writer needs to be created before the async logic to use the right XML writer factory
+            ComplexWriter complexWriter = complexWriterFactory.getComplexWriter(format);
+            futures.add(CompletableFuture.supplyAsync(
+                    () -> complexExporter.fetchAndWriteComplexes(complexesAcsChunk, complexWriter),
+                    executor));
+        }
+        return futures;
+    }
+
+    private String combineAllSerialisedContent(
+            SerialisedComplexesWriter serialisedComplexesWriter,
+            List<CompletableFuture<String>> serialisedContentFutures) throws IOException, ExecutionException, InterruptedException {
+
+        serialisedComplexesWriter.start();
+        for (CompletableFuture<String> future : serialisedContentFutures) {
+            String serialisedContent = future.get();
+            serialisedComplexesWriter.writeSerialisedContent(serialisedContent);
+        }
+        serialisedComplexesWriter.end();
+        return serialisedComplexesWriter.getString();
+
+    }
+
+    private boolean isQueryASingleId(String query) {
+        return (query.startsWith("EBI-") || query.startsWith("CPX-")) && query.trim().split(" ").length == 1;
+    }
+}

src/main/java/uk/ac/ebi/intact/service/complex/ws/ComplexExporter.java

@@ -0,0 +1,45 @@
+package uk.ac.ebi.intact.service.complex.ws;
+
+import lombok.AllArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import uk.ac.ebi.intact.jami.dao.IntactDao;
+import uk.ac.ebi.intact.jami.model.extension.IntactComplex;
+import uk.ac.ebi.intact.service.complex.ws.writer.ComplexWriter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+@Component
+@AllArgsConstructor
+public class ComplexExporter {
+
+    private final IntactDao intactDao;
+
+    @Transactional(readOnly = true, propagation = Propagation.REQUIRED, value = "jamiTransactionManager")
+    public String fetchAndWriteComplexes(Collection<String> complexesAcs,
+                                         ComplexWriter complexWriter) {
+
+        List<IntactComplex> complexes = fetchComplexes(complexesAcs);
+        return complexWriter.writeComplexes(complexes);
+    }
+
+    private List<IntactComplex> fetchComplexes(Collection<String> complexesAcs) {
+        List<IntactComplex> complexes = new ArrayList<>(complexesAcs.size());
+        for (String complexAc : complexesAcs) {
+            IntactComplex complex;
+            if (complexAc.startsWith("EBI-")) {
+                complex = intactDao.getComplexDao().getByAc(complexAc);
+            } else {
+                complex = intactDao.getComplexDao().getLatestComplexVersionByComplexAc(complexAc);
+            }
+
+            if (complex != null) {
+                complexes.add(complex);
+            }
+        }
+        return complexes;
+    }
+}