summaryrefslogtreecommitdiffstats
path: root/archiva-modules/archiva-scheduler
diff options
context:
space:
mode:
authorOlivier Lamy <olamy@apache.org>2013-11-07 05:21:39 +0000
committerOlivier Lamy <olamy@apache.org>2013-11-07 05:21:39 +0000
commit222ddd5b7d8f3f2798623ba9a6cf08c091a517f2 (patch)
tree5dea4e2c7063fbf39ddbf0804e05468d1f2449ac /archiva-modules/archiva-scheduler
parent609b3d7e9847d5df64f8417f8a0a68e1232d4873 (diff)
downloadarchiva-222ddd5b7d8f3f2798623ba9a6cf08c091a517f2.tar.gz
archiva-222ddd5b7d8f3f2798623ba9a6cf08c091a517f2.zip
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
Diffstat (limited to 'archiva-modules/archiva-scheduler')
-rw-r--r--archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml5
-rw-r--r--archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java224
-rw-r--r--archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/test/resources/log4j2-test.xml4
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>