use zero copy transfer for huge maven index

git-svn-id: https://svn.apache.org/repos/asf/archiva/trunk@1539520 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 222 insertions, 11 deletions

diff --git a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml
index 39d95b9f1..66ea8c1ed 100644
--- a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml
+++ b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml
@@ -72,6 +72,10 @@
       </exclusions>
     </dependency>
     <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpasyncclient</artifactId>
+    </dependency>
+    <dependency>
       <groupId>org.springframework</groupId>
       <artifactId>spring-context</artifactId>
     </dependency>
@@ -79,6 +83,7 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
+
     <dependency>
       <groupId>org.eclipse.jetty</groupId>
       <artifactId>jetty-server</artifactId>
diff --git a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java
index b0bea81c0..a25473300 100644
--- a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java
+++ b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java
@@ -26,7 +26,29 @@ import org.apache.archiva.proxy.common.WagonFactory;
 import org.apache.archiva.proxy.common.WagonFactoryException;
 import org.apache.archiva.proxy.common.WagonFactoryRequest;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.time.StopWatch;
+import org.apache.http.HttpException;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.ClientProtocolException;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.nio.ContentDecoder;
+import org.apache.http.nio.ContentEncoder;
+import org.apache.http.nio.IOControl;
+import org.apache.http.nio.client.methods.ZeroCopyConsumer;
+import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
+import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
+import org.apache.http.protocol.HttpContext;
 import org.apache.maven.index.context.IndexingContext;
 import org.apache.maven.index.updater.IndexUpdateRequest;
 import org.apache.maven.index.updater.IndexUpdater;
@@ -56,8 +78,13 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.security.Principal;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 /**
  * @author Olivier Lamy
@@ -170,6 +197,23 @@ public class DownloadRemoteIndexTask
             wagon.connect( new Repository( this.remoteRepository.getId(), baseIndexUrl ), authenticationInfo,
                            proxyInfo );
 
+            //---------------------------------------------
+
+            HttpAsyncClientBuilder builder = HttpAsyncClientBuilder.create();
+
+            if ( this.networkProxy != null )
+            {
+                HttpHost httpHost = new HttpHost( this.networkProxy.getHost(), this.networkProxy.getPort() );
+                builder = builder.setProxy( httpHost );
+            }
+
+            if ( this.remoteRepository.getUserName() != null )
+            {
+                BasicCredentialsProvider basicCredentialsProvider = new BasicCredentialsProvider();
+                basicCredentialsProvider.setCredentials( AuthScope.ANY, new UsernamePasswordCredentials(
+                    this.remoteRepository.getUserName(), this.remoteRepository.getPassword() ) );
+            }
+
             File indexDirectory = indexingContext.getIndexDirectoryFile();
             if ( !indexDirectory.exists() )
             {
@@ -178,6 +222,12 @@ public class DownloadRemoteIndexTask
 
             ResourceFetcher resourceFetcher =
                 new WagonResourceFetcher( log, tempIndexDirectory, wagon, remoteRepository );
+            CloseableHttpAsyncClient closeableHttpAsyncClient = builder.build();
+            closeableHttpAsyncClient.start();
+            resourceFetcher =
+                new ZeroCopyResourceFetcher( log, tempIndexDirectory, remoteRepository, closeableHttpAsyncClient,
+                                             baseIndexUrl );
+
             IndexUpdateRequest request = new IndexUpdateRequest( indexingContext, resourceFetcher );
             request.setForceFullUpdate( this.fullDownload );
             request.setLocalIndexCacheDir( indexCacheDirectory );
@@ -360,29 +410,183 @@ public class DownloadRemoteIndexTask
             }
         }
 
-        // FIXME remove crappy copy/paste
-        protected String addParameters( String path, RemoteRepository remoteRepository )
+    }
+
+    private static class ZeroCopyResourceFetcher
+        implements ResourceFetcher
+    {
+
+        Logger log;
+
+        File tempIndexDirectory;
+
+        final RemoteRepository remoteRepository;
+
+        CloseableHttpAsyncClient httpclient;
+
+        String baseIndexUrl;
+
+        private ZeroCopyResourceFetcher( Logger log, File tempIndexDirectory, RemoteRepository remoteRepository,
+                                         CloseableHttpAsyncClient httpclient, String baseIndexUrl )
+        {
+            this.log = log;
+            this.tempIndexDirectory = tempIndexDirectory;
+            this.remoteRepository = remoteRepository;
+            this.httpclient = httpclient;
+            this.baseIndexUrl = baseIndexUrl;
+        }
+
+        public void connect( String id, String url )
+            throws IOException
         {
-            if ( remoteRepository.getExtraParameters().isEmpty() )
+            //no op
+        }
+
+        public void disconnect()
+            throws IOException
+        {
+            // no op
+        }
+
+        public InputStream retrieve( final String name )
+            throws IOException, FileNotFoundException
+        {
+
+            log.info( "index update retrieve file, name:{}", name );
+            File file = new File( tempIndexDirectory, name );
+            if ( file.exists() )
             {
-                return path;
+                file.delete();
             }
+            file.deleteOnExit();
+
+            ZeroCopyConsumer<File> consumer = new ZeroCopyConsumer<File>( file )
+            {
 
-            boolean question = false;
+                @Override
+                protected File process( final HttpResponse response, final File file, final ContentType contentType )
+                    throws Exception
+                {
+                    if ( response.getStatusLine().getStatusCode() != HttpStatus.SC_OK )
+                    {
+                        throw new ClientProtocolException( "Upload failed: " + response.getStatusLine() );
+                    }
+                    return file;
+                }
 
-            StringBuilder res = new StringBuilder( path == null ? "" : path );
+                @Override
+                protected void onContentReceived( ContentDecoder decoder, IOControl ioctrl )
+                    throws IOException
+                {
+                    log.debug( "onContentReceived" );
+                    super.onContentReceived( decoder, ioctrl );
+                }
+            };
+            URL targetUrl = new URL( this.remoteRepository.getUrl() );
+            final HttpHost targetHost = new HttpHost( targetUrl.getHost(), targetUrl.getPort() );
 
-            for ( Map.Entry<String, String> entry : remoteRepository.getExtraParameters().entrySet() )
+            Future<File> httpResponseFuture = httpclient.execute( new HttpAsyncRequestProducer()
             {
-                if ( !question )
+                @Override
+                public HttpHost getTarget()
+                {
+                    return targetHost;
+                }
+
+                @Override
+                public HttpRequest generateRequest()
+                    throws IOException, HttpException
+                {
+                    StringBuilder url = new StringBuilder( baseIndexUrl );
+                    if ( !StringUtils.endsWith( baseIndexUrl, "/" ) )
+                    {
+                        url.append( '/' );
+                    }
+                    HttpGet httpGet = new HttpGet( url.append( addParameters( name, remoteRepository ) ).toString() );
+                    return httpGet;
+                }
+
+                @Override
+                public void produceContent( ContentEncoder encoder, IOControl ioctrl )
+                    throws IOException
+                {
+                    // no op
+                }
+
+                @Override
+                public void requestCompleted( HttpContext context )
+                {
+                    // no op
+                }
+
+                @Override
+                public void failed( Exception ex )
+                {
+                    log.error( "http request failed", ex );
+                }
+
+                @Override
+                public boolean isRepeatable()
+                {
+                    return true;
+                }
+
+                @Override
+                public void resetRequest()
+                    throws IOException
+                {
+                    // no op
+                }
+
+                @Override
+                public void close()
+                    throws IOException
                 {
-                    res.append( '?' ).append( entry.getKey() ).append( '=' ).append( entry.getValue() );
+                    // no op
                 }
+            }, consumer, null );
+            try
+            {
+                file = httpResponseFuture.get( this.remoteRepository.getTimeout(), TimeUnit.SECONDS );
+            }
+            catch ( InterruptedException e )
+            {
+                throw new IOException( e.getMessage(), e );
+            }
+            catch ( ExecutionException e )
+            {
+                throw new IOException( e.getMessage(), e );
+            }
+            catch ( TimeoutException e )
+            {
+                throw new IOException( e.getMessage(), e );
             }
+            return new FileInputStream( file );
+        }
 
-            return res.toString();
+    }
+
+    // FIXME remove crappy copy/paste
+    protected static String addParameters( String path, RemoteRepository remoteRepository )
+    {
+        if ( remoteRepository.getExtraParameters().isEmpty() )
+        {
+            return path;
+        }
+
+        boolean question = false;
+
+        StringBuilder res = new StringBuilder( path == null ? "" : path );
+
+        for ( Map.Entry<String, String> entry : remoteRepository.getExtraParameters().entrySet() )
+        {
+            if ( !question )
+            {
+                res.append( '?' ).append( entry.getKey() ).append( '=' ).append( entry.getValue() );
+            }
         }
 
+        return res.toString();
     }
 
 
diff --git a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/test/resources/log4j2-test.xml b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/test/resources/log4j2-test.xml
index 740df697e..6e3ef0ff3 100644
--- a/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/test/resources/log4j2-test.xml
+++ b/archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/test/resources/log4j2-test.xml
@@ -30,7 +30,9 @@
     <logger name="org.springframework" level="error"/>
 
     <logger name="org.apache.archiva.scheduler.indexing" level="debug"/>
-
+    <!--
+    <logger name="org.apache.http" level="debug"/>
+    -->
     <root level="info">
       <appender-ref ref="console"/>
     </root>