gCubeSystem
/
data-transfer-service
16
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

git-svn-id: http://svn.research-infrastructures.eu/public/d4science/gcube/branches/data-transfer/data-transfer-service/2.0@157503 82a268e6-3cf1-43bd-a215-b396298e98cf

This commit is contained in:
Fabio Sinibaldi 2017-10-23 15:41:23 +00:00
parent d3c941ba65
commit 14dd18e3da
2 changed files with 36 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
pom.xml
View File

@ -49,11 +49,7 @@
<artifactId>decompress-archive-plugin</artifactId> <artifactId>decompress-archive-plugin</artifactId>
<version>[1.0.0-SNAPSHOT,2.0.0-SNAPSHOT)</version> <version>[1.0.0-SNAPSHOT,2.0.0-SNAPSHOT)</version>
</dependency> </dependency>
<dependency>
<groupId>org.gcube.data.transfer</groupId>
<artifactId>sis-geotk-plugin</artifactId>
<version>LATEST</version>
</dependency>
<dependency> <dependency>

43
src/main/java/org/gcube/data/transfer/service/transfers/engine/impl/AbstractTicketHandler.java
View File

@ -9,6 +9,10 @@ import java.io.InputStream;
import java.io.OutputStream; import java.io.OutputStream;
import java.net.HttpURLConnection; import java.net.HttpURLConnection;
import java.net.URL; import java.net.URL;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.regex.Matcher; import java.util.regex.Matcher;
@ -40,11 +44,19 @@ public abstract class AbstractTicketHandler {
PersistenceProvider persistenceProvider; PersistenceProvider persistenceProvider;
PluginManager pluginManager; PluginManager pluginManager;
private MessageDigest md;
public AbstractTicketHandler(PersistenceProvider persProv,PluginManager plugMan, TransferTicket ticket) { public AbstractTicketHandler(PersistenceProvider persProv,PluginManager plugMan, TransferTicket ticket) {
this.persistenceProvider=persProv; this.persistenceProvider=persProv;
this.pluginManager=plugMan; this.pluginManager=plugMan;
this.ticket=ticket; this.ticket=ticket;
try {
md = MessageDigest.getInstance("SHA1");
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException("Unable to initialize",e);
}
} }
protected void onStep(String msg,double progress,Status status,long transferredBytes){ protected void onStep(String msg,double progress,Status status,long transferredBytes){
@ -120,13 +132,16 @@ public abstract class AbstractTicketHandler {
throw new ManagedException("Cannot save file in host"); throw new ManagedException("Cannot save file in host");
} }
transferStream(is, out); String checksum=transferStream(is, out);
completedTransfer=true; completedTransfer=true;
// IF TRANSFER FAILS, EXCEPTIONS AR THROWN // IF TRANSFER FAILS, EXCEPTIONS AR THROWN
log.debug("Completed transfer to {} , moving to destination {} ",tempFile.getAbsolutePath(),destination.getAbsolutePath()); log.debug("Completed transfer to {} [ SHA1 : {}]. moving to destination {} ",tempFile.getAbsolutePath(),checksum,destination.getAbsolutePath());
tempFile.renameTo(destination); Files.copy(tempFile.toPath(), destination.toPath(),StandardCopyOption.REPLACE_EXISTING);
Files.deleteIfExists(tempFile.toPath());
log.debug("Moved. Size is [temp : {} , dest : {}] ",tempFile.length(),destination.length());
//Plugin execution //Plugin execution
if(ticket.getPluginInvocations()!=null){ if(ticket.getPluginInvocations()!=null){
for(PluginInvocation invocation:ticket.getPluginInvocations()){ for(PluginInvocation invocation:ticket.getPluginInvocations()){
@ -185,25 +200,37 @@ public abstract class AbstractTicketHandler {
return getTicket(); return getTicket();
} }
private void transferStream(InputStream in, OutputStream out) throws ManagedException{
private String transferStream(InputStream in, OutputStream out) throws ManagedException{
md.reset();
long receivedTotal=0l; long receivedTotal=0l;
try{ try{
byte[] internalBuf=new byte[1024]; byte[] internalBuf=new byte[1024];
int received=0; int received=0;
while ((received=in.read(internalBuf))!=-1){ while ((received=in.read(internalBuf))!=-1){
md.update(internalBuf, 0, received);
out.write(internalBuf,0,received); out.write(internalBuf,0,received);
receivedTotal+=received; receivedTotal+=received;
onStep("Transferring",0d,Status.TRANSFERRING,receivedTotal); onStep("Transferring",0d,Status.TRANSFERRING,receivedTotal);
} }
out.flush(); out.flush();
byte[] mdbytes = md.digest();
//convert the byte to hex format
StringBuffer sb = new StringBuffer("");
for (int i = 0; i < mdbytes.length; i++) {
sb.append(Integer.toString((mdbytes[i] & 0xff) + 0x100, 16).substring(1));
}
log.debug("Completed transfer phase for ticket ID {}. Transferred {} bytes. ",ticket.getId(),receivedTotal);
return sb.toString();
}catch(IOException e){ }catch(IOException e){
log.debug("Unable to read from source",e); log.debug("Unable to read from source",e);
throw new ManagedException("Unable to read from source."); throw new ManagedException("Unable to read from source.");
} }
log.debug("Completed transfer phase for ticket ID {}. Transferred {} bytes. ",ticket.getId(),receivedTotal);
} }
private InputStream getInputStream() throws ManagedException{ private InputStream getInputStream() throws ManagedException{