Menu
▾
▴
[Unicore-cvs-commits] SF.net SVN: unicore:[11171] workflow/trunk/workflowservice/src
|
From: <bsc...@us...> - 2011-09-15 14:51:07
|
Revision: 11171
http://unicore.svn.sourceforge.net/unicore/?rev=11171&view=rev
Author: bschuller
Date: 2011-09-15 14:50:57 +0000 (Thu, 15 Sep 2011)
Log Message:
-----------
allow simple way to specify pattern in chunked for loop
Modified Paths:
--------------
workflow/trunk/workflowservice/src/main/java/eu/unicore/workflow/pe/util/InsertVariablesFilter.java
workflow/trunk/workflowservice/src/test/java/eu/unicore/workflow/pe/util/TestInsertVariablesFilter.java
Modified: workflow/trunk/workflowservice/src/main/java/eu/unicore/workflow/pe/util/InsertVariablesFilter.java
===================================================================
--- workflow/trunk/workflowservice/src/main/java/eu/unicore/workflow/pe/util/InsertVariablesFilter.java 2011-09-15 14:27:36 UTC (rev 11170)
+++ workflow/trunk/workflowservice/src/main/java/eu/unicore/workflow/pe/util/InsertVariablesFilter.java 2011-09-15 14:50:57 UTC (rev 11171)
@@ -31,81 +31,81 @@
public class InsertVariablesFilter {
private static final Logger logger=LogUtil.getLogger(LogUtil.SERVICES,InsertVariablesFilter.class);
-
+
private final ProcessVariables context;
-
+
public InsertVariablesFilter(ProcessVariables context){
this.context=context;
}
-
+
/**
* 1) extract inputs where source is the workflow context and target
* the environment. Create environment variables from these. <br/>
* 2) replace occurences of "${VARIABLENAME}" in stagein/stageout
* by the appropriate values
* 3) replace occurences of "${VARIABLENAME}" in job name
- *
*/
public WorkAssignmentDocument filter(WorkAssignmentDocument wa) {
try{
JobDefinitionDocument job=JobDefinitionDocument.Factory.newInstance();
job.setJobDefinition(wa.getWorkAssignment().getWork().getJobDefinition());
+
+ JSDLUtils ju=new JSDLUtils(job);
+ replaceVariablesInJobName(ju);
+
Boolean chunked=(Boolean)context.get(ChunkedFileIterator.PV_IS_CHUNKED);
if(Boolean.TRUE.equals(chunked)){
handleChunkedInputs(job);
}
-
- JSDLUtils ju=new JSDLUtils(job);
+
POSIXApplicationDocument pApp=ju.getPOSIXApplicationCopy();
if(pApp==null){
pApp=POSIXApplicationDocument.Factory.newInstance();
pApp.addNewPOSIXApplication();
}
-
+
replaceVariablesInStagingSections(ju, pApp);
-
- replaceVariablesInJobName(ju);
-
+
removeDeprecatedContextReferences(pApp);
-
+
replaceVariablesInEnvironmentValues(pApp);
-
+
ju.setPOSIXApplication(pApp);
wa.getWorkAssignment().getWork().setJobDefinition(ju.getJsdl().getJobDefinition());
return wa;
-
+
}catch(Exception e){
logger.warn("Can't process workassignment document",e);
return wa;
}
}
-
+
/**
* replace occurences of ${VARNAME} in the data staging sections
* @param pApp
*/
protected void replaceVariablesInStagingSections(JSDLUtils ju, POSIXApplicationDocument pApp)
- throws UnsupportedEncodingException, URISyntaxException{
-
+ throws UnsupportedEncodingException, URISyntaxException{
+
for(Output t: ju.getOutputs()){
String source=t.getSource();
source=expandVariables(source);
t.setSource(source);
-
+
//this is really an URI, so it needs to be decoded and later encoded
String target=URLDecoder.decode(t.getTarget(),"UTF-8");
target=expandVariables(target);
t.setTarget(URIUtils.encodeAll(target));
}
-
+
for(Input t: ju.getInputs()){
//this is really an URI, so it needs to be decoded and later encoded
String source=URLDecoder.decode(t.getSource(),"UTF-8");
source=expandVariables(source);
t.setSource(URIUtils.encodeAll(source));
-
+
String target=t.getTarget();
t.setTarget(expandVariables(target));
}
@@ -123,8 +123,8 @@
ju.getJsdl().getJobDefinition().getJobDescription().getJobIdentification().setJobName(jobName);
}catch(Exception ignored){}
}
-
-
+
+
/**
* remove the "context://" stuff used in the old WF engine
* @param pApp
@@ -137,7 +137,7 @@
}
}
}
-
+
/**
* replace occurences of ${VARNAME} in the values of the environment variables
* @param pApp
@@ -159,18 +159,18 @@
String iteratorName=(String)context.get(ChunkedFileIterator.PV_ITERATOR_NAME);
String key="${"+iteratorName+"}";
String pattern=(String)context.get(ChunkedFileIterator.PV_FILENAME_FORMAT);
-
+
List<DataStagingType>results=new ArrayList<DataStagingType>();
-
+
for(DataStagingType dst: job.getJobDefinition().getJobDescription().getDataStagingArray()){
if(dst.getSource()==null)continue;
-
+
String source=URLDecoder.decode(dst.getSource().getURI(),"UTF-8");
if(!source.contains(key)){
results.add(dst);
continue;
}
-
+
//else need to clone it several timers
for(int i=1;i<=thisChunkSize;i++){
DataStagingType clone=(DataStagingType)dst.copy();
@@ -181,26 +181,33 @@
clone.getSource().setURI(newSource);
results.add(clone);
}
-
+
}
int r=results.size();
if(r>0){
job.getJobDefinition().getJobDescription().setDataStagingArray(results.toArray(new DataStagingType[r]));
}
}
-
+
protected String getFormattedFilename(String filename, String pattern, int index){
- //split off extension
- int lastDot=filename.lastIndexOf(".");
- String name=filename;
- String ext="";
- if(lastDot>-1){
- ext=filename.substring(lastDot);
- name=filename.substring(0, lastDot);
+ //check if the filename contains a "*" indicating how the user wants to pattern to be
+ if(filename.contains("*")){
+ String formattedIndex=MessageFormat.format("{0,number,0000}", index);
+ return filename.replaceAll("\\*", formattedIndex);
}
- return MessageFormat.format(pattern, index, name, ext);
+ else{
+ //split off extension
+ int lastDot=filename.lastIndexOf(".");
+ String name=filename;
+ String ext="";
+ if(lastDot>-1){
+ ext=filename.substring(lastDot);
+ name=filename.substring(0, lastDot);
+ }
+ return MessageFormat.format(pattern, index, name, ext);
+ }
}
-
+
protected String expandVariables(String var){
if(var.contains("${")){
for(Object o: context.keySet()){
Modified: workflow/trunk/workflowservice/src/test/java/eu/unicore/workflow/pe/util/TestInsertVariablesFilter.java
===================================================================
--- workflow/trunk/workflowservice/src/test/java/eu/unicore/workflow/pe/util/TestInsertVariablesFilter.java 2011-09-15 14:27:36 UTC (rev 11170)
+++ workflow/trunk/workflowservice/src/test/java/eu/unicore/workflow/pe/util/TestInsertVariablesFilter.java 2011-09-15 14:50:57 UTC (rev 11171)
@@ -2,7 +2,6 @@
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
-import java.net.URLEncoder;
import org.chemomentum.common.util.jsdl.JSDLUtils;
import org.chemomentum.workassignment.xmlbeans.WorkAssignmentDocument;
@@ -134,4 +133,32 @@
assert ju.getInputs().get(1).getTarget().equals("infile_0002.txt");
}
+
+ public void testManipulateStageInDefinitionsUsingSimplePattern(){
+ ProcessVariables vars=new ProcessVariables();
+ vars.put(ChunkedFileIterator.PV_IS_CHUNKED, Boolean.TRUE);
+ vars.put(ChunkedFileIterator.PV_ITERATOR_NAME, "ForEachIterator");
+ vars.put(ChunkedFileIterator.PV_THIS_CHUNK_SIZE, Integer.valueOf(2));
+ vars.put(ChunkedFileIterator.PV_FILENAME+"_1", "foo_1");
+ vars.put(ChunkedFileIterator.PV_FILENAME+"_2", "foo_2");
+
+ InsertVariablesFilter filter=new InsertVariablesFilter(vars);
+
+ WABuilder builder=new WABuilder();
+ builder.runApplication("Date");
+ builder.withInputFile("${ForEachIterator}", "infile_*.txt");
+ WorkAssignmentDocument doc=builder.getWorkAssignmentDocument();
+
+ WorkAssignmentDocument filtered=filter.filter(doc);
+ JobDefinitionDocument jdd=JobDefinitionDocument.Factory.newInstance();
+ jdd.setJobDefinition(filtered.getWorkAssignment().getWork().getJobDefinition());
+ JSDLUtils ju=new JSDLUtils(jdd);
+
+ assert 2==ju.getInputs().size();
+ assert ju.getInputs().get(0).getSource().equals("foo_1");
+ assert ju.getInputs().get(0).getTarget().equals("infile_0001.txt");
+
+ assert ju.getInputs().get(1).getSource().equals("foo_2");
+ assert ju.getInputs().get(1).getTarget().equals("infile_0002.txt");
+ }
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|