You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2013/02/06 00:09:38 UTC
svn commit: r1442791 - in
/airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler:
GFacHandlerException.java GramDirectorySetupHandler.java
GridFTPInputHandler.java GridFTPOutputHandler.java
Author: lahiru
Date: Tue Feb 5 23:09:37 2013
New Revision: 1442791
URL: http://svn.apache.org/viewvc?rev=1442791&view=rev
Log:
adding new set of handlers.
Added:
airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GramDirectorySetupHandler.java
airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPInputHandler.java
airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPOutputHandler.java
Modified:
airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GFacHandlerException.java
Modified: airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GFacHandlerException.java
URL: http://svn.apache.org/viewvc/airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GFacHandlerException.java?rev=1442791&r1=1442790&r2=1442791&view=diff
==============================================================================
--- airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GFacHandlerException.java (original)
+++ airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GFacHandlerException.java Tue Feb 5 23:09:37 2013
@@ -21,6 +21,9 @@
package org.apache.airavata.gfac.handler;
+import org.apache.airavata.gfac.context.JobExecutionContext;
+import org.apache.airavata.gfac.notification.events.ExecutionFailEvent;
+
public class GFacHandlerException extends Exception{
public GFacHandlerException(String s) {
super(s);
@@ -29,4 +32,27 @@ public class GFacHandlerException extend
public GFacHandlerException(String s, Throwable throwable) {
super(s, throwable);
}
+ public GFacHandlerException(String message, Throwable cause,JobExecutionContext context) {
+ super(message, cause);
+ sendFaultNotification(message,context,new Exception(cause));
+ }
+
+ public GFacHandlerException(String message, JobExecutionContext context) {
+ super(message);
+ sendFaultNotification(message,context,new Exception(message));
+ }
+
+ public GFacHandlerException(String message, JobExecutionContext context,Exception e,String... additionExceptiondata) {
+ super(message);
+ sendFaultNotification(message,context,e, additionExceptiondata);
+ }
+
+ private void sendFaultNotification(String message,
+ JobExecutionContext executionContext, Exception e,
+ String... additionalExceptiondata) {
+ if (additionalExceptiondata==null || additionalExceptiondata.length==0){
+ additionalExceptiondata=new String[]{message,e.getLocalizedMessage()};
+ }
+ executionContext.getNotifier().publish(new ExecutionFailEvent(e));
+ }
}
Added: airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GramDirectorySetupHandler.java
URL: http://svn.apache.org/viewvc/airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GramDirectorySetupHandler.java?rev=1442791&view=auto
==============================================================================
--- airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GramDirectorySetupHandler.java (added)
+++ airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GramDirectorySetupHandler.java Tue Feb 5 23:09:37 2013
@@ -0,0 +1,91 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+*/
+package org.apache.airavata.gfac.handler;
+
+import org.apache.airavata.gfac.ToolsException;
+import org.apache.airavata.gfac.context.GSISecurityContext;
+import org.apache.airavata.gfac.context.JobExecutionContext;
+import org.apache.airavata.gfac.external.GridFtp;
+import org.apache.airavata.gfac.provider.GFacProviderException;
+import org.apache.airavata.gfac.utils.GFacUtils;
+import org.apache.airavata.gfac.utils.GramJobSubmissionListener;
+import org.apache.airavata.schemas.gfac.ApplicationDeploymentDescriptionType;
+import org.apache.airavata.schemas.gfac.GlobusHostType;
+import org.ietf.jgss.GSSCredential;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+public class GramDirectorySetupHandler implements GFacHandler {
+ private static final Logger log = LoggerFactory.getLogger(GramJobSubmissionListener.class);
+
+ public void invoke(JobExecutionContext jobExecutionContext) throws GFacHandlerException {
+ GlobusHostType host = (GlobusHostType) jobExecutionContext.getApplicationContext().getHostDescription().getType();
+ ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType();
+ GridFtp ftp = new GridFtp();
+
+ try {
+ GSISecurityContext gssContext = new GSISecurityContext(jobExecutionContext.getGFacConfiguration());
+ GSSCredential gssCred = gssContext.getGssCredentails();
+ String[] hostgridFTP = host.getGridFTPEndPointArray();
+ if (hostgridFTP == null || hostgridFTP.length == 0) {
+ hostgridFTP = new String[]{host.getHostAddress()};
+ }
+ boolean success = false;
+ GFacHandlerException pe = null;// = new ProviderException("");
+ for (String endpoint : host.getGridFTPEndPointArray()) {
+ try {
+
+ URI tmpdirURI = GFacUtils.createGsiftpURI(endpoint, app.getScratchWorkingDirectory());
+ URI workingDirURI = GFacUtils.createGsiftpURI(endpoint, app.getStaticWorkingDirectory());
+ URI inputURI = GFacUtils.createGsiftpURI(endpoint, app.getInputDataDirectory());
+ URI outputURI = GFacUtils.createGsiftpURI(endpoint, app.getOutputDataDirectory());
+
+ log.info("Host FTP = " + hostgridFTP[0]);
+ log.info("temp directory = " + tmpdirURI);
+ log.info("Working directory = " + workingDirURI);
+ log.info("Input directory = " + inputURI);
+ log.info("Output directory = " + outputURI);
+
+ ftp.makeDir(tmpdirURI, gssCred);
+ ftp.makeDir(workingDirURI, gssCred);
+ ftp.makeDir(inputURI, gssCred);
+ ftp.makeDir(outputURI, gssCred);
+
+ success = true;
+ break;
+ } catch (URISyntaxException e) {
+ pe = new GFacHandlerException("URI is malformatted:" + e.getMessage(), e, jobExecutionContext);
+
+ } catch (ToolsException e) {
+ pe = new GFacHandlerException(e.getMessage(), e, jobExecutionContext);
+ }
+ }
+ if (success == false) {
+ throw pe;
+ }
+ } catch (SecurityException e) {
+ throw new GFacHandlerException(e.getMessage(), e, jobExecutionContext);
+ }
+ }
+}
Added: airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPInputHandler.java
URL: http://svn.apache.org/viewvc/airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPInputHandler.java?rev=1442791&view=auto
==============================================================================
--- airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPInputHandler.java (added)
+++ airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPInputHandler.java Tue Feb 5 23:09:37 2013
@@ -0,0 +1,113 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+*/
+package org.apache.airavata.gfac.handler;
+
+import org.apache.airavata.commons.gfac.type.ActualParameter;
+import org.apache.airavata.commons.gfac.type.MappingFactory;
+import org.apache.airavata.gfac.ToolsException;
+import org.apache.airavata.gfac.context.GSISecurityContext;
+import org.apache.airavata.gfac.context.JobExecutionContext;
+import org.apache.airavata.gfac.context.MessageContext;
+import org.apache.airavata.gfac.external.GridFtp;
+import org.apache.airavata.gfac.provider.GFacProviderException;
+import org.apache.airavata.gfac.utils.GFacUtils;
+import org.apache.airavata.schemas.gfac.ApplicationDeploymentDescriptionType;
+import org.apache.airavata.schemas.gfac.GlobusHostType;
+import org.apache.airavata.schemas.gfac.URIArrayType;
+import org.apache.airavata.schemas.gfac.URIParameterType;
+import org.ietf.jgss.GSSCredential;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+public class GridFTPInputHandler implements GFacHandler{
+ public void invoke(JobExecutionContext jobExecutionContext) throws GFacHandlerException {
+ MessageContext inputNew = new MessageContext();
+ try {
+ MessageContext input = jobExecutionContext.getInMessageContext();
+ Set<String> parameters = input.getParameters().keySet();
+ for (String paramName:parameters) {
+ ActualParameter actualParameter = (ActualParameter) input.getParameters().get(paramName);
+ String paramValue = MappingFactory.toString(actualParameter);
+ //TODO: Review this with type
+ if ("URI".equals(actualParameter.getType().getType().toString())) {
+ ((URIParameterType) actualParameter.getType()).setValue(stageInputFiles(jobExecutionContext, paramValue));
+ } else if ("URIArray".equals(actualParameter.getType().getType().toString())) {
+ List<String> split = Arrays.asList(paramValue.split(","));
+ List<String> newFiles = new ArrayList<String>();
+ for (String paramValueEach : split) {
+ newFiles.add(stageInputFiles(jobExecutionContext, paramValueEach));
+ }
+ ((URIArrayType) actualParameter.getType()).setValueArray(newFiles.toArray(new String[newFiles.size()]));
+ }
+ inputNew.getParameters().put(paramName, actualParameter);
+ }
+ } catch (Exception e) {
+// jobExecutionContext.getExecutionContext().getNotifier().executionFail(jobExecutionContext,e,"Error during Input File staging");
+ throw new GFacHandlerException("Error while input File Staging", jobExecutionContext, e, e.getLocalizedMessage());
+ }
+ jobExecutionContext.setInMessageContext(inputNew);
+ }
+ private static String stageInputFiles(JobExecutionContext jobExecutionContext, String paramValue) throws URISyntaxException, SecurityException, ToolsException, IOException {
+ URI gridftpURL;
+ gridftpURL = new URI(paramValue);
+ GlobusHostType host = (GlobusHostType) jobExecutionContext.getApplicationContext().getHostDescription().getType();
+ ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType();
+ GridFtp ftp = new GridFtp();
+ URI destURI = null;
+ GSISecurityContext gssContext = new GSISecurityContext(jobExecutionContext.getGFacConfiguration());
+ GSSCredential gssCred = gssContext.getGssCredentails();
+
+ for (String endpoint : host.getGridFTPEndPointArray()) {
+ URI inputURI = GFacUtils.createGsiftpURI(endpoint, app.getInputDataDirectory());
+ String fileName = new File(gridftpURL.getPath()).getName();
+ String s = inputURI.getPath() + File.separator + fileName;
+ //if user give a url just to refer an endpoint, not a web resource we are not doing any transfer
+ if (fileName != null && !"".equals(fileName)) {
+ destURI = GFacUtils.createGsiftpURI(endpoint, s);
+ if (paramValue.startsWith("gsiftp")) {
+ ftp.uploadFile(gridftpURL, destURI, gssCred);
+ } else if (paramValue.startsWith("file")) {
+ String localFile = paramValue.substring(paramValue.indexOf(":") + 1, paramValue.length());
+ ftp.uploadFile(destURI, gssCred, new FileInputStream(localFile));
+ } else if (paramValue.startsWith("http")) {
+ ftp.uploadFile(destURI,
+ gssCred, (gridftpURL.toURL().openStream()));
+ } else {
+ //todo throw exception telling unsupported protocol
+ return paramValue;
+ }
+ } else {
+ // When the given input is not a web resource but a URI type input, then we don't do any transfer just keep the same value as it isin the input
+ return paramValue;
+ }
+ }
+ System.out.println(destURI.getPath());
+ return destURI.getPath();
+ }
+}
Added: airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPOutputHandler.java
URL: http://svn.apache.org/viewvc/airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPOutputHandler.java?rev=1442791&view=auto
==============================================================================
--- airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPOutputHandler.java (added)
+++ airavata/trunk/modules/gfac-core/src/main/java/org/apache/airavata/gfac/handler/GridFTPOutputHandler.java Tue Feb 5 23:09:37 2013
@@ -0,0 +1,222 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+*/
+package org.apache.airavata.gfac.handler;
+
+import org.apache.airavata.commons.gfac.type.ActualParameter;
+import org.apache.airavata.commons.gfac.type.MappingFactory;
+import org.apache.airavata.gfac.ToolsException;
+import org.apache.airavata.gfac.context.GSISecurityContext;
+import org.apache.airavata.gfac.context.JobExecutionContext;
+import org.apache.airavata.gfac.context.MessageContext;
+import org.apache.airavata.gfac.external.GridFtp;
+import org.apache.airavata.gfac.notification.events.ExecutionFailEvent;
+import org.apache.airavata.gfac.provider.GFacProviderException;
+import org.apache.airavata.gfac.utils.GFacUtils;
+import org.apache.airavata.gfac.utils.GramJobSubmissionListener;
+import org.apache.airavata.gfac.utils.OutputUtils;
+import org.apache.airavata.schemas.gfac.*;
+import org.ietf.jgss.GSSCredential;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.*;
+
+public class GridFTPOutputHandler implements GFacHandler{
+ private static final Logger log = LoggerFactory.getLogger(GramJobSubmissionListener.class);
+
+ public void invoke(JobExecutionContext jobExecutionContext) throws GFacHandlerException {
+ GlobusHostType host = (GlobusHostType) jobExecutionContext.getApplicationContext().getHostDescription().getType();
+ ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType();
+ GridFtp ftp = new GridFtp();
+ File localStdErrFile = null;
+ Map<String, ActualParameter> stringMap = null;
+ try {
+ GSISecurityContext gssContext = new GSISecurityContext(jobExecutionContext.getGFacConfiguration());
+ GSSCredential gssCred = gssContext.getGssCredentails();
+
+ String[] hostgridFTP = host.getGridFTPEndPointArray();
+ if (hostgridFTP == null || hostgridFTP.length == 0) {
+ hostgridFTP = new String[]{host.getHostAddress()};
+ }
+ for (String endpoint : host.getGridFTPEndPointArray()) {
+ try {
+ /*
+ * Read Stdout and Stderror
+ */
+ URI stdoutURI = GFacUtils.createGsiftpURI(endpoint, app.getStandardOutput());
+ URI stderrURI = GFacUtils.createGsiftpURI(endpoint, app.getStandardError());
+
+ log.info("STDOUT:" + stdoutURI.toString());
+ log.info("STDERR:" + stderrURI.toString());
+
+ File logDir = new File("./service_logs");
+ if (!logDir.exists()) {
+ logDir.mkdir();
+ }
+
+ String timeStampedServiceName = GFacUtils.createUniqueNameForService(jobExecutionContext
+ .getServiceName());
+ File localStdOutFile = File.createTempFile(timeStampedServiceName, "stdout");
+ localStdErrFile = File.createTempFile(timeStampedServiceName, "stderr");
+
+ String stdout = ftp.readRemoteFile(stdoutURI, gssCred, localStdOutFile);
+ String stderr = ftp.readRemoteFile(stderrURI, gssCred, localStdErrFile);
+ Map<String, Object> output = jobExecutionContext.getOutMessageContext().getParameters();
+ Set<String> keys = output.keySet();
+ for (String paramName : keys) {
+ ActualParameter actualParameter = (ActualParameter) output.get(paramName);
+ if ("URIArray".equals(actualParameter.getType().getType().toString())) {
+ URI outputURI = GFacUtils.createGsiftpURI(endpoint, app.getOutputDataDirectory());
+ List<String> outputList = ftp.listDir(outputURI, gssCred);
+ String[] valueList = outputList.toArray(new String[outputList.size()]);
+ ((URIArrayType) actualParameter.getType()).setValueArray(valueList);
+ stringMap = new HashMap<String, ActualParameter>();
+ stringMap.put(paramName, actualParameter);
+ }
+ if ("StringArray".equals(actualParameter.getType().getType().toString())) {
+ String[] valueList = OutputUtils.parseStdoutArray(stdout, paramName);
+ ((StringArrayType) actualParameter.getType()).setValueArray(valueList);
+ stringMap = new HashMap<String, ActualParameter>();
+ stringMap.put(paramName, actualParameter);
+ } else {
+ // This is to handle exception during the output parsing.
+ stringMap = OutputUtils.fillOutputFromStdout(jobExecutionContext, stdout, stderr);
+ }
+ }
+ if (stringMap == null || stringMap.isEmpty()) {
+ jobExecutionContext.getNotifier().publish(new ExecutionFailEvent(new Throwable("Empty Output returned from the Application, Double check the application" +
+ "and ApplicationDescriptor output Parameter Names")));
+// GFacProviderException exception = new GFacProviderException("Gram provider: Error creating job output", jobExecutionContext);
+// jobExecutionContext.getExecutionContext().getNotifier().executionFail(jobExecutionContext,exception,exception.getLocalizedMessage());
+// throw exception;
+ }
+ //todo check the workflow context header and run the stateOutputFiles method to stage the output files in to a user defined location
+ stageOutputFiles(jobExecutionContext,app.getOutputDataDirectory());
+ } catch (ToolsException e) {
+ throw new GFacHandlerException(e.getMessage(), jobExecutionContext, e, readLastLinesofStdOut(localStdErrFile.getPath(), 20));
+ } catch (URISyntaxException e) {
+ throw new GFacHandlerException("URI is malformatted:" + e.getMessage(), jobExecutionContext, e, readLastLinesofStdOut(localStdErrFile.getPath(), 20));
+ } catch (NullPointerException e) {
+ throw new GFacHandlerException("Output is not produced in stdout:" + e.getMessage(), jobExecutionContext, e, readLastLinesofStdOut(localStdErrFile.getPath(), 20));
+ }
+ }
+ } catch (Exception e) {
+// jobExecutionContext.getExecutionContext().getNotifier().executionFail(jobExecutionContext,e,readLastLinesofStdOut(localStdErrFile.getPath(), 20));
+ throw new GFacHandlerException(e.getMessage(), jobExecutionContext, e, readLastLinesofStdOut(localStdErrFile.getPath(), 20));
+ }
+ }
+
+ private static String readLastLinesofStdOut(String path, int count) {
+ StringBuffer buffer = new StringBuffer();
+ FileInputStream in = null;
+ try {
+ in = new FileInputStream(path);
+ } catch (FileNotFoundException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ BufferedReader br = new BufferedReader(new InputStreamReader(in));
+ List<String> strLine = new ArrayList<String>();
+ String tmp = null;
+ int numberofLines = 0;
+ try {
+ while ((tmp = br.readLine()) != null) {
+ strLine.add(tmp);
+ numberofLines++;
+ }
+ } catch (IOException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ if (numberofLines > count) {
+ for (int i = numberofLines - count; i < numberofLines; i++) {
+ buffer.append(strLine.get(i));
+ buffer.append("\n");
+ }
+ } else {
+ for (int i = 0; i < numberofLines; i++) {
+ buffer.append(strLine.get(i));
+ buffer.append("\n");
+ }
+ }
+ try {
+ in.close();
+ } catch (IOException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ return buffer.toString();
+ }
+
+ private static void stageOutputFiles(JobExecutionContext jobExecutionContext, String outputFileStagingPath) throws GFacProviderException {
+ MessageContext outputNew = new MessageContext();
+ MessageContext output = jobExecutionContext.getOutMessageContext();
+ Map<String, Object> parameters = output.getParameters();
+ for (String paramName : parameters.keySet()) {
+ ActualParameter actualParameter = (ActualParameter) parameters
+ .get(paramName);
+ //TODO: Review this with type
+ GlobusHostType host = (GlobusHostType) jobExecutionContext.getApplicationContext().getHostDescription().getType();
+ GridFtp ftp = new GridFtp();
+
+ GSISecurityContext gssContext = new GSISecurityContext(jobExecutionContext.getGFacConfiguration());
+ GSSCredential gssCred = null;
+ try {
+ gssCred = gssContext.getGssCredentails();
+ } catch (SecurityException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ try {
+ if ("URI".equals(actualParameter.getType().getType().toString())) {
+ for (String endpoint : host.getGridFTPEndPointArray()) {
+ ((URIParameterType) actualParameter.getType()).setValue(doStaging(outputFileStagingPath,
+ MappingFactory.toString(actualParameter), ftp, gssCred, endpoint));
+ }
+ } else if ("URIArray".equals(actualParameter.getType().getType().toString())) {
+ List<String> split = Arrays.asList(MappingFactory.toString(actualParameter).split(","));
+ List<String> newFiles = new ArrayList<String>();
+ for (String endpoint : host.getGridFTPEndPointArray()) {
+ for (String paramValueEach : split) {
+ newFiles.add(doStaging(outputFileStagingPath, paramValueEach, ftp, gssCred, endpoint));
+ }
+ ((URIArrayType) actualParameter.getType()).setValueArray(newFiles.toArray(new String[newFiles.size()]));
+ }
+
+ }
+ } catch (URISyntaxException e) {
+ throw new GFacProviderException(e.getMessage(), e, jobExecutionContext);
+ } catch (ToolsException e) {
+ throw new GFacProviderException(e.getMessage(), e, jobExecutionContext);
+ }
+ outputNew.getParameters().put(paramName, actualParameter);
+ }
+ jobExecutionContext.setOutMessageContext(outputNew);
+ }
+
+ private static String doStaging(String outputFileStagingPath, String paramValue, GridFtp ftp, GSSCredential gssCred, String endpoint) throws URISyntaxException, ToolsException {
+ URI srcURI = GFacUtils.createGsiftpURI(endpoint, paramValue);
+ String fileName = new File(srcURI.getPath()).getName();
+ File outputFile = new File(outputFileStagingPath + File.separator + fileName);
+ ftp.readRemoteFile(srcURI,
+ gssCred, outputFile);
+ return outputFileStagingPath + File.separator + fileName;
+ }
+}