Apache
/
archiva
mirror of https://github.com/apache/archiva.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

back to use wagon for remote download index to prevent unit test failure 

git-svn-id: https://svn.apache.org/repos/asf/archiva/trunk@1541818 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Olivier Lamy 2013-11-14 04:55:51 +00:00
parent c230bdbe41
commit 0bd43cf6f8
7 changed files with 124 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
archiva-modules/archiva-scheduler/archiva-scheduler-indexing/pom.xml
View File

@ -204,19 +204,14 @@
org.springframework*;version="[3,4)",
org.apache.archiva.redback.components.taskqueue*,
org.apache.maven.index*,
org.apache.http,
org.apache.http.auth,
org.apache.http.client,
org.apache.http.client.methods,
org.apache.http.concurrent,
org.apache.http.entity,
org.apache.http.impl.client,
org.apache.http.impl.nio.client,
org.apache.http.nio,
org.apache.http.nio.client.methods,
org.apache.http.nio.protocol,
org.apache.http.protocol,
org.apache.http.impl.nio.codecs,
org.apache.maven.wagon,
org.apache.maven.wagon.authentication,
org.apache.maven.wagon.authorization,
org.apache.maven.wagon.events,
org.apache.maven.wagon.providers.http,
org.apache.maven.wagon.proxy,
org.apache.maven.wagon.repository,
org.apache.maven.wagon.resource,
org.slf4j;resolution:=optional
</Import-Package>
</instructions>

338
archiva-modules/archiva-scheduler/archiva-scheduler-indexing/src/main/java/org/apache/archiva/scheduler/indexing/DownloadRemoteIndexTask.java
View File

@ -23,34 +23,29 @@ import org.apache.archiva.admin.model.beans.NetworkProxy;
import org.apache.archiva.admin.model.beans.RemoteRepository;
import org.apache.archiva.admin.model.remote.RemoteRepositoryAdmin;
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.HttpEntity;
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.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.impl.nio.codecs.LengthDelimitedDecoder;
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.protocol.HttpContext;
import org.apache.maven.index.context.IndexingContext;
import org.apache.maven.index.updater.IndexUpdateRequest;
import org.apache.maven.index.updater.IndexUpdater;
import org.apache.maven.index.updater.ResourceFetcher;
import org.apache.maven.wagon.ConnectionException;
import org.apache.maven.wagon.ResourceDoesNotExistException;
import org.apache.maven.wagon.StreamWagon;
import org.apache.maven.wagon.TransferFailedException;
import org.apache.maven.wagon.Wagon;
import org.apache.maven.wagon.authentication.AuthenticationException;
import org.apache.maven.wagon.authentication.AuthenticationInfo;
import org.apache.maven.wagon.authorization.AuthorizationException;
import org.apache.maven.wagon.events.TransferEvent;
import org.apache.maven.wagon.events.TransferListener;
import org.apache.maven.wagon.providers.http.AbstractHttpClientWagon;
import org.apache.maven.wagon.providers.http.HttpConfiguration;
import org.apache.maven.wagon.providers.http.HttpMethodConfiguration;
import org.apache.maven.wagon.proxy.ProxyInfo;
import org.apache.maven.wagon.repository.Repository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -59,15 +54,10 @@ import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.net.MalformedURLException;
import java.net.URL;
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
@ -140,9 +130,6 @@ public class DownloadRemoteIndexTask
tempIndexDirectory.deleteOnExit();
String baseIndexUrl = indexingContext.getIndexUpdateUrl();
URL indexUrl = new URL( baseIndexUrl );
/*
String wagonProtocol = new URL( this.remoteRepository.getUrl() ).getProtocol();
final StreamWagon wagon = (StreamWagon) wagonFactory.getWagon(
@ -182,38 +169,6 @@ public class DownloadRemoteIndexTask
}
wagon.connect( new Repository( this.remoteRepository.getId(), baseIndexUrl ), authenticationInfo,
proxyInfo );
*/
//---------------------------------------------
HttpAsyncClientBuilder builder = HttpAsyncClientBuilder.create();
BasicCredentialsProvider basicCredentialsProvider = new BasicCredentialsProvider();
if ( this.networkProxy != null )
{
HttpHost httpHost = new HttpHost( this.networkProxy.getHost(), this.networkProxy.getPort() );
builder = builder.setProxy( httpHost );
if ( this.networkProxy.getUsername() != null )
{
basicCredentialsProvider.setCredentials(
new AuthScope( this.networkProxy.getHost(), this.networkProxy.getPort(), null, null ),
new UsernamePasswordCredentials( this.networkProxy.getUsername(),
this.networkProxy.getPassword() ) );
}
}
if ( this.remoteRepository.getUserName() != null )
{
basicCredentialsProvider.setCredentials(
new AuthScope( indexUrl.getHost(), indexUrl.getPort(), null, null ),
new UsernamePasswordCredentials( this.remoteRepository.getUserName(),
this.remoteRepository.getPassword() ) );
}
builder = builder.setDefaultCredentialsProvider( basicCredentialsProvider );
File indexDirectory = indexingContext.getIndexDirectoryFile();
if ( !indexDirectory.exists() )
@ -221,36 +176,44 @@ public class DownloadRemoteIndexTask
indexDirectory.mkdirs();
}
CloseableHttpAsyncClient closeableHttpAsyncClient = builder.build();
closeableHttpAsyncClient.start();
ResourceFetcher resourceFetcher =
new ZeroCopyResourceFetcher( log, tempIndexDirectory, remoteRepository, closeableHttpAsyncClient,
baseIndexUrl );
new WagonResourceFetcher( log, tempIndexDirectory, wagon, remoteRepository );
IndexUpdateRequest request = new IndexUpdateRequest( indexingContext, resourceFetcher );
request.setForceFullUpdate( this.fullDownload );
request.setLocalIndexCacheDir( indexCacheDirectory );
this.indexUpdater.fetchAndUpdateIndex( request );
stopWatch.stop();
log.info( "time update index from remote for repository {}: {} s", this.remoteRepository.getId(),
( stopWatch.getTime() / 1000 ) );
// index packing optionnal ??
//IndexPackingRequest indexPackingRequest =
// new IndexPackingRequest( indexingContext, indexingContext.getIndexDirectoryFile() );
//indexPacker.packIndex( indexPackingRequest );
indexingContext.updateTimestamp( true );
stopWatch.stop();
log.info( "time update index from remote for repository {}: {} s", this.remoteRepository.getId(),
( stopWatch.getTime() / 1000 ) );
}
catch ( MalformedURLException e )
{
log.error( e.getMessage(), e );
throw new RuntimeException( e.getMessage(), e );
}
catch ( WagonFactoryException e )
{
log.error( e.getMessage(), e );
throw new RuntimeException( e.getMessage(), e );
}
catch ( ConnectionException e )
{
log.error( e.getMessage(), e );
throw new RuntimeException( e.getMessage(), e );
}
catch ( AuthenticationException e )
{
log.error( e.getMessage(), e );
throw new RuntimeException( e.getMessage(), e );
}
catch ( IOException e )
{
log.error( e.getMessage(), e );
@ -266,7 +229,7 @@ public class DownloadRemoteIndexTask
deleteDirectoryQuiet( tempIndexDirectory );
this.runningRemoteDownloadIds.remove( this.remoteRepository.getId() );
}
log.info( "end download remote index for remote repository {}", this.remoteRepository.getId() );
log.info( "end download remote index for remote repository " + this.remoteRepository.getId() );
}
private void deleteDirectoryQuiet( File f )
@ -281,8 +244,9 @@ public class DownloadRemoteIndexTask
}
}
private static class ZeroCopyConsumerListener
extends ZeroCopyConsumer
private static final class DownloadListener
implements TransferListener
{
private Logger log = LoggerFactory.getLogger( getClass() );
@ -290,87 +254,49 @@ public class DownloadRemoteIndexTask
private long startTime;
private long totalLength = 0;
private int totalLength = 0;
//private long currentLength = 0;
private ZeroCopyConsumerListener( File file, String resourceName )
throws FileNotFoundException
public void transferInitiated( TransferEvent transferEvent )
{
super( file );
this.resourceName = resourceName;
startTime = System.currentTimeMillis();
resourceName = transferEvent.getResource().getName();
log.debug( "initiate transfer of {}", resourceName );
}
@Override
protected File process( final HttpResponse response, final File file, final ContentType contentType )
throws Exception
public void transferStarted( TransferEvent transferEvent )
{
if ( response.getStatusLine().getStatusCode() != HttpStatus.SC_OK )
{
throw new ClientProtocolException( "Download failed: " + response.getStatusLine() );
this.totalLength = 0;
resourceName = transferEvent.getResource().getName();
log.info( "start transfer of {}", transferEvent.getResource().getName() );
}
public void transferProgress( TransferEvent transferEvent, byte[] buffer, int length )
{
log.debug( "transfer of {} : {}/{}", transferEvent.getResource().getName(), buffer.length, length );
this.totalLength += length;
}
public void transferCompleted( TransferEvent transferEvent )
{
resourceName = transferEvent.getResource().getName();
long endTime = System.currentTimeMillis();
log.info( "end of transfer file {} {} kb: {}s", resourceName, this.totalLength / 1024,
( endTime - startTime ) / 1000 );
return file;
log.info( "end of transfer file {} {} kb: {}s", transferEvent.getResource().getName(),
this.totalLength / 1024, ( endTime - startTime ) / 1000 );
}
@Override
protected void onContentReceived( ContentDecoder decoder, IOControl ioControl )
throws IOException
public void transferError( TransferEvent transferEvent )
{
if ( decoder instanceof LengthDelimitedDecoder )
log.info( "error of transfer file {}: {}", transferEvent.getResource().getName(),
transferEvent.getException().getMessage(), transferEvent.getException() );
}
public void debug( String message )
{
LengthDelimitedDecoder ldl = LengthDelimitedDecoder.class.cast( decoder );
long len = getLen( ldl );
if ( len > -1 )
{
log.debug( "transfer of {} : {}/{}", resourceName, len / 1024, this.totalLength / 1024 );
log.debug( "transfer debug {}", message );
}
}
super.onContentReceived( decoder, ioControl );
}
@Override
protected void onResponseReceived( HttpResponse response )
{
this.startTime = System.currentTimeMillis();
super.onResponseReceived( response );
this.totalLength = response.getEntity().getContentLength();
log.info( "start transfer of {}, contentLength: {}", resourceName, this.totalLength / 1024 );
}
@Override
protected void onEntityEnclosed( HttpEntity entity, ContentType contentType )
throws IOException
{
super.onEntityEnclosed( entity, contentType );
}
private long getLen( LengthDelimitedDecoder ldl )
{
try
{
Field lenField = LengthDelimitedDecoder.class.getDeclaredField( "len" );
lenField.setAccessible( true );
long len = (Long) lenField.get( ldl );
return len;
}
catch ( NoSuchFieldException e )
{
log.debug( e.getMessage(), e );
return -1;
}
catch ( IllegalAccessException e )
{
log.debug( e.getMessage(), e );
return -1;
}
}
}
private static class ZeroCopyResourceFetcher
private static class WagonResourceFetcher
implements ResourceFetcher
{
@ -378,20 +304,17 @@ public class DownloadRemoteIndexTask
File tempIndexDirectory;
final RemoteRepository remoteRepository;
Wagon wagon;
CloseableHttpAsyncClient httpclient;
RemoteRepository remoteRepository;
String baseIndexUrl;
private ZeroCopyResourceFetcher( Logger log, File tempIndexDirectory, RemoteRepository remoteRepository,
CloseableHttpAsyncClient httpclient, String baseIndexUrl )
private WagonResourceFetcher( Logger log, File tempIndexDirectory, Wagon wagon,
RemoteRepository remoteRepository )
{
this.log = log;
this.tempIndexDirectory = tempIndexDirectory;
this.wagon = wagon;
this.remoteRepository = remoteRepository;
this.httpclient = httpclient;
this.baseIndexUrl = baseIndexUrl;
}
public void connect( String id, String url )
@ -406,10 +329,11 @@ public class DownloadRemoteIndexTask
// no op
}
public InputStream retrieve( final String name )
throws IOException
public InputStream retrieve( String name )
throws IOException, FileNotFoundException
{
try
{
log.info( "index update retrieve file, name:{}", name );
File file = new File( tempIndexDirectory, name );
if ( file.exists() )
@ -417,98 +341,27 @@ public class DownloadRemoteIndexTask
file.delete();
}
file.deleteOnExit();
ZeroCopyConsumer<File> consumer = new ZeroCopyConsumerListener( file, name );
URL targetUrl = new URL( this.baseIndexUrl );
final HttpHost targetHost = new HttpHost( targetUrl.getHost(), targetUrl.getPort() );
Future<File> httpResponseFuture = httpclient.execute( new HttpAsyncRequestProducer()
{
@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 )
{
log.debug( "requestCompleted" );
}
@Override
public void failed( Exception ex )
{
log.error( "http request failed", ex );
}
@Override
public boolean isRepeatable()
{
log.debug( "isRepeatable" );
return true;
}
@Override
public void resetRequest()
throws IOException
{
log.debug( "resetRequest" );
}
@Override
public void close()
throws IOException
{
log.debug( "close" );
}
}, consumer, null );
try
{
int timeOut = this.remoteRepository.getRemoteDownloadTimeout();
file = timeOut > 0 ? httpResponseFuture.get( timeOut, TimeUnit.SECONDS ) : httpResponseFuture.get();
}
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 );
}
wagon.get( addParameters( name, this.remoteRepository ), file );
return new FileInputStream( file );
}
catch ( AuthorizationException e )
{
throw new IOException( e.getMessage(), e );
}
catch ( TransferFailedException e )
{
throw new IOException( e.getMessage(), e );
}
catch ( ResourceDoesNotExistException e )
{
FileNotFoundException fnfe = new FileNotFoundException( e.getMessage() );
fnfe.initCause( e );
throw fnfe;
}
}
// FIXME remove crappy copy/paste
protected static String addParameters( String path, RemoteRepository remoteRepository )
protected String addParameters( String path, RemoteRepository remoteRepository )
{
if ( remoteRepository.getExtraParameters().isEmpty() )
{
@ -530,7 +383,8 @@ public class DownloadRemoteIndexTask
return res.toString();
}
}
}

2
archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/AbstractDownloadTest.java → archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/remotedownload/AbstractDownloadTest.java
View File

@ -1,4 +1,4 @@
package org.apache.archiva;
package org.apache.archiva.remotedownload;
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file

2
archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/DownloadArtifactsTest.java → archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/remotedownload/DownloadArtifactsTest.java
View File

@ -1,4 +1,4 @@
package org.apache.archiva;
package org.apache.archiva.remotedownload;
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file

3
archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/DownloadMergedIndexNonDefaultPathTest.java → archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/remotedownload/DownloadMergedIndexNonDefaultPathTest.java
View File

@ -1,4 +1,4 @@
package org.apache.archiva;
package org.apache.archiva.remotedownload;
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@ -186,6 +186,7 @@ public class DownloadMergedIndexNonDefaultPathTest
while ( !repositoriesService.getRunningRemoteDownloadIds().getStrings().isEmpty() )
{
Thread.sleep( 500 );
log.debug( "still running remote download" );
}
SearchService searchService = getSearchService();

2
archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/DownloadMergedIndexTest.java → archiva-modules/archiva-web/archiva-web-common/src/test/java/org/apache/archiva/remotedownload/DownloadMergedIndexTest.java
View File

@ -1,4 +1,4 @@
package org.apache.archiva;
package org.apache.archiva.remotedownload;
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file

2
archiva-modules/archiva-web/archiva-web-common/src/test/resources/log4j2-test.xml
View File

@ -35,6 +35,8 @@
<logger name="org.apache.archiva.scheduler.indexing" level="debug"/>
<logger name="org.apache.archiva.remotedownload" level="debug"/>
<root level="info">
<appender-ref ref="console"/>
</root>