Author: bfoster
Date: Tue May 1 22:36:17 2012
New Revision: 1332872
URL: http://svn.apache.org/viewvc?rev=1332872&view=rev
Log:
- Add file staging support to CAS-PGE
--------------
OODT-443
Added:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java
(with props)
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java
(with props)
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java
(with props)
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java
(with props)
oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/
oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java
(with props)
Modified:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETask.java
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETaskInstance.java
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/PgeConfig.java
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/metadata/PgeTaskMetKeys.java
Modified: oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETask.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETask.java?rev=1332872&r1=1332871&r2=1332872&view=diff
==============================================================================
--- oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETask.java (original)
+++ oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETask.java Tue May
1 22:36:17 2012
@@ -14,12 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
-
package org.apache.oodt.cas.pge;
+//JDK static imports
+import static
org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createPGETaskInstance;
+
//JDK imports
import java.io.File;
+import java.util.logging.Logger;
//OODT imports
import org.apache.oodt.cas.workflow.structs.WorkflowTaskConfiguration;
@@ -28,18 +30,16 @@ import org.apache.oodt.cas.metadata.Meta
import org.apache.oodt.cas.metadata.SerializableMetadata;
/**
- *
- * @author mattmann
- * @author bfoster
- * @version $Revision$
- *
- * <p>
* Runs a {@link PGETaskInstance} given {@link Metadata} and a
- * {@link WorkflowTaskConfiguration}
- * </p>.
+ * {@link WorkflowTaskConfiguration}.
+ *
+ * @author mattmann (Chris Mattmann)
+ * @author bfoster (Brian Foster)
*/
public class PGETask {
+ private static final Logger LOGGER =
Logger.getLogger(PGETask.class.getName());
+
private Metadata metadata;
private WorkflowTaskConfiguration wftConfig;
@@ -52,8 +52,8 @@ public class PGETask {
public void run(String pgeTaskInstanceClasspath)
throws InstantiationException, IllegalAccessException,
ClassNotFoundException, WorkflowTaskInstanceException {
- PGETaskInstance pgeTaskInst = (PGETaskInstance) Class.forName(
- pgeTaskInstanceClasspath).newInstance();
+ PGETaskInstance pgeTaskInst = createPGETaskInstance(
+ pgeTaskInstanceClasspath, LOGGER);
pgeTaskInst.run(this.metadata, this.wftConfig);
}
Modified:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETaskInstance.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETaskInstance.java?rev=1332872&r1=1332871&r2=1332872&view=diff
==============================================================================
--- oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETaskInstance.java
(original)
+++ oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/PGETaskInstance.java
Tue May 1 22:36:17 2012
@@ -23,6 +23,7 @@ import static org.apache.oodt.cas.pge.me
import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.CRAWLER_CRAWL_FOR_DIRS;
import static org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.CRAWLER_RECUR;
import static org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.DUMP_METADATA;
+import static org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.FILE_STAGER;
import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.INGEST_CLIENT_TRANSFER_SERVICE_FACTORY;
import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.INGEST_FILE_MANAGER_URL;
import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.LOG_FILENAME_PATTERN;
@@ -36,6 +37,10 @@ import static org.apache.oodt.cas.pge.me
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.CONF_FILE_BUILD;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.CRAWLING;
import static org.apache.oodt.cas.pge.metadata.PgeTaskStatus.RUNNING_PGE;
+import static
org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createConfigFilePropertyAdder;
+import static
org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createFileStager;
+import static
org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createPgeConfigBuilder;
+import static
org.apache.oodt.cas.pge.util.GenericPgeObjectFactory.createSciPgeConfigFileWriter;
//JDK imports
import java.io.File;
@@ -61,10 +66,11 @@ import org.apache.oodt.cas.metadata.Seri
import org.apache.oodt.cas.pge.config.DynamicConfigFile;
import org.apache.oodt.cas.pge.config.OutputDir;
import org.apache.oodt.cas.pge.config.PgeConfig;
-import org.apache.oodt.cas.pge.config.PgeConfigBuilder;
import org.apache.oodt.cas.pge.config.XmlFilePgeConfigBuilder;
import org.apache.oodt.cas.pge.metadata.PgeMetadata;
import org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys;
+import org.apache.oodt.cas.pge.staging.FileManagerFileStager;
+import org.apache.oodt.cas.pge.staging.FileStager;
import org.apache.oodt.cas.pge.writers.SciPgeConfigFileWriter;
import org.apache.oodt.cas.workflow.metadata.CoreMetKeys;
import org.apache.oodt.cas.workflow.structs.WorkflowTaskConfiguration;
@@ -118,6 +124,7 @@ public class PGETaskInstance implements
createOuputDirsIfRequested();
updateStatus(CONF_FILE_BUILD.getWorkflowStatusName());
createSciPgeConfigFiles();
+ stageFiles();
// Run the PGE.
runPge();
@@ -135,7 +142,7 @@ public class PGETaskInstance implements
}
protected void updateStatus(String status) throws Exception {
- logger.log(Level.INFO, "Updating status to workflow as [" + status +
"]");
+ logger.info("Updating status to workflow as [" + status + "]");
if (!wm.updateWorkflowInstanceStatus(workflowInstId, status)) {
throw new Exception(
"Failed to update workflow status : client returned false");
@@ -169,8 +176,7 @@ public class PGETaskInstance implements
protected PgeMetadata createPgeMetadata(Metadata dynMetadata,
WorkflowTaskConfiguration config) throws Exception {
- logger.log(Level.INFO,
- "Converting workflow configuration to static metadata...");
+ logger.info("Converting workflow configuration to static metadata...");
Metadata staticMetadata = new Metadata();
for (Object objKey : config.getProperties().keySet()) {
String key = (String) objKey;
@@ -180,19 +186,19 @@ public class PGETaskInstance implements
Splitter.on(",").trimResults()
.omitEmptyStrings()
.split(config.getProperty(key)));
- logger.log(Level.FINEST, "Adding static metadata: key = [" + key
+ logger.finest("Adding static metadata: key = [" + key
+ "] value = " + values);
staticMetadata.addMetadata(key, values);
} else {
String value = config.getProperty(key);
- logger.log(Level.FINEST, "Adding static metadata: key = [" + key
+ logger.finest("Adding static metadata: key = [" + key
+ "] value = [" + value + "]");
staticMetadata.addMetadata(key, value);
}
}
- logger.log(Level.INFO, "Loading workflow context metadata...");
+ logger.info("Loading workflow context metadata...");
for (String key : dynMetadata.getAllKeys()) {
- logger.log(Level.FINEST,
+ logger.finest(
"Adding dynamic metadata: key = [" + key + "] value = "
+ dynMetadata.getAllMetadata(key));
}
@@ -200,15 +206,15 @@ public class PGETaskInstance implements
}
protected PgeConfig createPgeConfig() throws Exception {
- logger.log(Level.INFO, "Create PgeConfig...");
+ logger.info("Create PgeConfig...");
String pgeConfigBuilderClass = pgeMetadata
.getMetadata(PGE_CONFIG_BUILDER);
if (pgeConfigBuilderClass != null) {
- logger.log(Level.INFO, "Using PgeConfigBuilder: " +
pgeConfigBuilderClass);
- return ((PgeConfigBuilder) Class.forName(pgeConfigBuilderClass)
- .newInstance()).build(pgeMetadata);
+ logger.info("Using PgeConfigBuilder: " + pgeConfigBuilderClass);
+ return createPgeConfigBuilder(pgeConfigBuilderClass, logger)
+ .build(pgeMetadata);
} else {
- logger.log(Level.INFO, "Using default PgeConfigBuilder: "
+ logger.info("Using default PgeConfigBuilder: "
+ XmlFilePgeConfigBuilder.class.getCanonicalName());
return new XmlFilePgeConfigBuilder().build(pgeMetadata);
}
@@ -216,7 +222,7 @@ public class PGETaskInstance implements
protected void runPropertyAdders() throws Exception {
try {
- logger.log(Level.INFO, "Loading/Running property adders...");
+ logger.info("Loading/Running property adders...");
List<String> propertyAdders = pgeMetadata
.getAllMetadata(PROPERTY_ADDERS);
if (propertyAdders != null) {
@@ -224,7 +230,7 @@ public class PGETaskInstance implements
runPropertyAdder(loadPropertyAdder(propertyAdder));
}
} else {
- logger.log(Level.INFO, "No property adders specified");
+ logger.info("No property adders specified");
}
} catch (Exception e) {
throw new Exception("Failed to instanciate/run Property Adders : "
@@ -234,14 +240,13 @@ public class PGETaskInstance implements
protected ConfigFilePropertyAdder loadPropertyAdder(
String propertyAdderClasspath) throws Exception {
- logger.log(Level.FINE, "Loading property adder: " +
propertyAdderClasspath);
- return (ConfigFilePropertyAdder) Class.forName(propertyAdderClasspath)
- .newInstance();
+ logger.fine("Loading property adder: " + propertyAdderClasspath);
+ return createConfigFilePropertyAdder(propertyAdderClasspath, logger);
}
protected void runPropertyAdder(ConfigFilePropertyAdder propAdder)
throws Exception {
- logger.log(Level.INFO, "Running property adder: "
+ logger.info("Running property adder: "
+ propAdder.getClass().getCanonicalName());
propAdder.addConfigProperties(pgeMetadata,
pgeConfig.getPropertyAdderCustomArgs());
@@ -250,14 +255,14 @@ public class PGETaskInstance implements
protected XmlRpcWorkflowManagerClient createWorkflowManagerClient()
throws Exception {
String url = pgeMetadata.getMetadata(WORKFLOW_MANAGER_URL);
- logger.log(Level.INFO, "Creating WorkflowManager client for url [" + url
+ "]");
+ logger.info("Creating WorkflowManager client for url [" + url + "]");
Validate.notNull(url, "Must specify " + WORKFLOW_MANAGER_URL);
return new XmlRpcWorkflowManagerClient(new URL(url));
}
protected String getWorkflowInstanceId() throws Exception {
String instanceId =
pgeMetadata.getMetadata(CoreMetKeys.WORKFLOW_INST_ID);
- logger.log(Level.INFO, "Workflow instanceId is [" + instanceId + "]");
+ logger.info("Workflow instanceId is [" + instanceId + "]");
Validate.notNull(instanceId, "Must specify "
+ CoreMetKeys.WORKFLOW_INST_ID);
return instanceId;
@@ -282,7 +287,7 @@ public class PGETaskInstance implements
}
protected void createExeDir() throws Exception {
- logger.log(Level.INFO, "Creating PGE execution working directory: ["
+ logger.info("Creating PGE execution working directory: ["
+ pgeConfig.getExeDir() + "]");
File executionDir = new File(pgeConfig.getExeDir());
if (!(executionDir.exists() || executionDir.mkdirs())) {
@@ -294,7 +299,7 @@ public class PGETaskInstance implements
protected void createOuputDirsIfRequested() throws Exception {
for (OutputDir outputDir : pgeConfig.getOuputDirs()) {
if (outputDir.isCreateBeforeExe()) {
- logger.log(Level.INFO, "Creating PGE file ouput directory: ["
+ logger.info("Creating PGE file ouput directory: ["
+ outputDir.getPath() + "]");
File dir = new File(outputDir.getPath());
if (!(dir.exists() || dir.mkdirs())) {
@@ -305,19 +310,42 @@ public class PGETaskInstance implements
}
}
+ protected void stageFiles() throws Exception {
+ if (pgeConfig.getFileStagingInfo() != null) {
+ FileStager fileStager = getFileStager();
+ logger.info("Starting file staging...");
+ fileStager.stageFiles(
+ pgeConfig.getFileStagingInfo(), pgeMetadata, logger);
+ } else {
+ logger.info("No files to stage.");
+ }
+ }
+
+ protected FileStager getFileStager() throws Exception {
+ String fileStagerClass = pgeMetadata.getMetadata(FILE_STAGER);
+ if (fileStagerClass != null) {
+ logger.info("Loading FileStager [" + fileStagerClass + "]");
+ return createFileStager(fileStagerClass, logger);
+ } else {
+ logger.info("Using default FileStager ["
+ + FileManagerFileStager.class.getCanonicalName() + "]");
+ return new FileManagerFileStager();
+ }
+ }
+
protected void createSciPgeConfigFiles() throws Exception {
- logger.log(Level.INFO, "Starting creation of science PGE files...");
+ logger.info("Starting creation of science PGE files...");
for (DynamicConfigFile dynamicConfigFile : pgeConfig
.getDynamicConfigFiles()) {
createSciPgeConfigFile(dynamicConfigFile);
}
- logger.log(Level.INFO, "Successfully wrote all science PGE files!");
+ logger.info("Successfully wrote all science PGE files!");
}
protected void createSciPgeConfigFile(DynamicConfigFile dynamicConfigFile)
throws Exception {
Validate.notNull(dynamicConfigFile, "dynamicConfigFile cannot be null");
- logger.log(Level.FINE, "Starting creation of science PGE file ["
+ logger.fine("Starting creation of science PGE file ["
+ dynamicConfigFile.getFilePath() + "]...");
// Create parent directory if it doesn't exist.
@@ -329,15 +357,14 @@ public class PGETaskInstance implements
}
// Load writer and write file.
- logger.log(Level.FINE, "Loading writer class for science PGE file ["
+ logger.fine("Loading writer class for science PGE file ["
+ dynamicConfigFile.getFilePath() + "]...");
- SciPgeConfigFileWriter writer = (SciPgeConfigFileWriter) Class.forName(
- dynamicConfigFile.getWriterClass()).newInstance();
- logger.log(Level.FINE, "Loaded writer [" +
writer.getClass().getCanonicalName()
+ SciPgeConfigFileWriter writer = createSciPgeConfigFileWriter(
+ dynamicConfigFile.getWriterClass(), logger);
+ logger.fine("Loaded writer [" + writer.getClass().getCanonicalName()
+ "] for science PGE file [" + dynamicConfigFile.getFilePath()
+ "]...");
- logger.log(Level.INFO,
- "Writing science PGE file [" + dynamicConfigFile.getFilePath()
+ logger.info("Writing science PGE file [" +
dynamicConfigFile.getFilePath()
+ "]...");
File configFile =
writer.createConfigFile(dynamicConfigFile.getFilePath(),
pgeMetadata.asMetadata(), dynamicConfigFile.getArgs());
@@ -348,8 +375,7 @@ public class PGETaskInstance implements
}
protected ScriptFile buildPgeRunScript() {
- logger.log(Level.FINE,
- "Creating PGE run script for shell [" + pgeConfig.getShellType()
+ logger.fine("Creating PGE run script for shell [" +
pgeConfig.getShellType()
+ "] with contents " + pgeConfig.getExeCmds());
ScriptFile sf = new ScriptFile(pgeConfig.getShellType());
sf.setCommands(pgeConfig.getExeCmds());
@@ -358,13 +384,13 @@ public class PGETaskInstance implements
protected File getScriptPath() {
File script = new File(pgeConfig.getExeDir(), getPgeScriptName());
- logger.log(Level.FINE, "Script file with be written to [" + script +
"]");
+ logger.fine("Script file with be written to [" + script + "]");
return script;
}
protected String getPgeScriptName() {
String pgeScriptName = "sciPgeExeScript_" +
pgeMetadata.getMetadata(NAME);
- logger.log(Level.FINE, "Generated script file name [" + pgeScriptName +
"]");
+ logger.fine("Generated script file name [" + pgeScriptName + "]");
return pgeScriptName;
}
@@ -372,7 +398,7 @@ public class PGETaskInstance implements
ScriptFile sf = null;
try {
long startTime = System.currentTimeMillis();
- logger.log(Level.INFO, "PGE start time [" + new Date(startTime) +
"]");
+ logger.info("PGE start time [" + new Date(startTime) + "]");
// create script to run
sf = buildPgeRunScript();
@@ -380,21 +406,21 @@ public class PGETaskInstance implements
// run script and evaluate whether success or failure
updateStatus(RUNNING_PGE.getWorkflowStatusName());
- logger.log(Level.INFO, "Starting execution of PGE...");
+ logger.info("Starting execution of PGE...");
if (!wasPgeSuccessful(ExecUtils.callProgram(
pgeConfig.getShellType() + " " + getScriptPath(), logger,
new File(pgeConfig.getExeDir()).getAbsoluteFile()))) {
throw new RuntimeException("Pge didn't finish successfully");
} else {
- logger.log(Level.INFO,
+ logger.info(
"Successfully completed running: '" + sf.getCommands() +
"'");
}
long endTime = System.currentTimeMillis();
- logger.log(Level.INFO, "PGE end time [" + new Date(startTime) + "]");
+ logger.info("PGE end time [" + new Date(startTime) + "]");
long runTime = endTime - startTime;
- logger.log(Level.INFO, "PGE runtime in millis [" + runTime + "]");
+ logger.info("PGE runtime in millis [" + runTime + "]");
pgeMetadata.replaceMetadata(PGE_RUNTIME, Long.toString(runTime));
@@ -410,7 +436,7 @@ public class PGETaskInstance implements
}
protected ProductCrawler createProductCrawler() throws Exception {
- logger.log(Level.INFO, "Configuring ProductCrawler...");
+ logger.info("Configuring ProductCrawler...");
AutoDetectProductCrawler crawler = new AutoDetectProductCrawler();
crawler.setMimeExtractorRepo(pgeMetadata.getMetadata(MIME_EXTRACTOR_REPO));
crawler.setClientTransferer(pgeMetadata
@@ -430,10 +456,10 @@ public class PGETaskInstance implements
.getMetadata(CRAWLER_CRAWL_FOR_DIRS)));
crawler.setNoRecur(!Boolean.parseBoolean(
pgeMetadata.getMetadata(CRAWLER_RECUR)));
- logger.log(Level.FINE,
+ logger.fine(
"Passing Workflow Metadata to CAS-Crawler as global metadata . .
.");
crawler.setGlobalMetadata(pgeMetadata.asMetadata(PgeMetadata.Type.DYNAMIC));
- logger.log(Level.FINE, "Created ProductCrawler ["
+ logger.fine("Created ProductCrawler ["
+ crawler.getClass().getCanonicalName() + "]");
return crawler;
}
@@ -450,7 +476,7 @@ public class PGETaskInstance implements
boolean attemptIngestAll = Boolean.parseBoolean(pgeMetadata
.getMetadata(ATTEMPT_INGEST_ALL));
for (File crawlDir : crawlDirs) {
- logger.log(Level.INFO, "Crawling for products in [" + crawlDir + "]");
+ logger.info("Crawling for products in [" + crawlDir + "]");
crawler.crawl(crawlDir);
if (!attemptIngestAll) {
verifyIngests(crawler);
@@ -462,7 +488,7 @@ public class PGETaskInstance implements
}
protected void verifyIngests(ProductCrawler crawler) throws Exception {
- logger.log(Level.INFO, "Verifying ingests successful...");
+ logger.info("Verifying ingests successful...");
boolean ingestsSuccess = true;
String exceptionMsg = "";
for (IngestStatus status : crawler.getIngestStatus()) {
@@ -473,7 +499,7 @@ public class PGETaskInstance implements
+ status.getResult() + "',msg='" + status.getMessage() +
"']";
ingestsSuccess = false;
} else if (!status.getResult().equals(IngestStatus.Result.SUCCESS)) {
- logger.log(Level.WARNING, "Product was not ingested [file='"
+ logger.warning("Product was not ingested [file='"
+ status.getProduct().getAbsolutePath() + "',result='"
+ status.getResult() + "',msg='" + status.getMessage() +
"']");
}
@@ -481,7 +507,7 @@ public class PGETaskInstance implements
if (!ingestsSuccess) {
throw new Exception(exceptionMsg);
} else {
- logger.log(Level.INFO, "Ingests were successful");
+ logger.info("Ingests were successful");
}
}
Added:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java?rev=1332872&view=auto
==============================================================================
---
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java
(added)
+++
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java
Tue May 1 22:36:17 2012
@@ -0,0 +1,81 @@
+/*
+ * 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.oodt.cas.pge.config;
+
+//JDK imports
+import java.util.List;
+import java.util.Set;
+
+//Google imports
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+/**
+ * Configuration information about which files should be staged and where.
+ *
+ * @author bfoster (Brian Foster)
+ */
+public class FileStagingInfo {
+
+ private String stagingDir;
+ private boolean forceStaging;
+ private Set<String> filePaths;
+ private Set<String> productIds;
+
+ public FileStagingInfo(String stagingDir) {
+ this(stagingDir, false);
+ }
+
+ public FileStagingInfo(String stagingDir, boolean forceStaging) {
+ this.stagingDir = stagingDir;
+ this.forceStaging = forceStaging;
+ filePaths = Sets.newHashSet();
+ productIds = Sets.newHashSet();
+ }
+
+ public void addFilePath(String filePath) {
+ filePaths.add(filePath);
+ }
+
+ public void addFilePaths(List<String> filePaths) {
+ filePaths.addAll(filePaths);
+ }
+
+ public List<String> getFilePaths() {
+ return Lists.newArrayList(filePaths);
+ }
+
+ public void addProductId(String productId) {
+ productIds.add(productId);
+ }
+
+ public void addProductIds(List<String> productIds) {
+ productIds.addAll(productIds);
+ }
+
+ public List<String> getProductIds() {
+ return Lists.newArrayList(productIds);
+ }
+
+ public String getStagingDir() {
+ return stagingDir;
+ }
+
+ public boolean isForceStaging() {
+ return forceStaging;
+ }
+}
Propchange:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/FileStagingInfo.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/PgeConfig.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/PgeConfig.java?rev=1332872&r1=1332871&r2=1332872&view=diff
==============================================================================
--- oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/PgeConfig.java
(original)
+++ oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/config/PgeConfig.java
Tue May 1 22:36:17 2012
@@ -36,6 +36,7 @@ public class PgeConfig {
private String exeDir;
private String shellType;
private List<String> exeCmds;
+ private FileStagingInfo fileStagingInfo;
public PgeConfig() {
shellType = "sh";
@@ -94,4 +95,12 @@ public class PgeConfig {
return propertyAdderCustomArgs != null ? propertyAdderCustomArgs
: new Object[0];
}
+
+ public void setFileStagingInfo(FileStagingInfo fileStagingInfo) {
+ this.fileStagingInfo = fileStagingInfo;
+ }
+
+ public FileStagingInfo getFileStagingInfo() {
+ return fileStagingInfo;
+ }
}
Modified:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/metadata/PgeTaskMetKeys.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/metadata/PgeTaskMetKeys.java?rev=1332872&r1=1332871&r2=1332872&view=diff
==============================================================================
---
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/metadata/PgeTaskMetKeys.java
(original)
+++
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/metadata/PgeTaskMetKeys.java
Tue May 1 22:36:17 2012
@@ -57,6 +57,14 @@ public enum PgeTaskMetKeys {
"PGETask/PropertyAdders",
"PGETask_PropertyAdderClasspath",
true),
+
+ /**
+ * {@link FileStager}s classpath.
+ */
+ FILE_STAGER(
+ "PGETask/FileStager",
+ "PGETask/FileStager"),
+
/**
* List of {@link ConfigFilePropertyAdder}s classpaths to be run.
*/
Added:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java?rev=1332872&view=auto
==============================================================================
---
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java
(added)
+++
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java
Tue May 1 22:36:17 2012
@@ -0,0 +1,94 @@
+/*
+ * 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.oodt.cas.pge.staging;
+
+//OODT static imports
+import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.QUERY_CLIENT_TRANSFER_SERVICE_FACTORY;
+import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.QUERY_FILE_MANAGER_URL;
+
+//OODT imports
+import org.apache.oodt.cas.filemgr.datatransfer.DataTransfer;
+import org.apache.oodt.cas.filemgr.datatransfer.RemoteDataTransferFactory;
+import org.apache.oodt.cas.filemgr.structs.Product;
+import org.apache.oodt.cas.filemgr.structs.Reference;
+import org.apache.oodt.cas.filemgr.util.GenericFileManagerObjectFactory;
+import org.apache.oodt.cas.pge.metadata.PgeMetadata;
+
+//Google imports
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+
+//JDK imports
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URL;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A {@link FileStager} which uses a FileManager {@link DataTransferer}.
+ *
+ * @author bfoster (Brian Foster)
+ */
+public class FileManagerFileStager extends FileStager {
+
+ @Override
+ public void stageFile(URI stageFile, File destDir,
+ PgeMetadata pgeMetadata, Logger logger) throws Exception {
+ DataTransfer dataTransferer = createDataTransfer(pgeMetadata, logger);
+ logger.log(Level.INFO, "Using DataTransfer ["
+ + dataTransferer.getClass().getCanonicalName() + "]");
+ setFileManagerUrl(dataTransferer, pgeMetadata, logger);
+ dataTransferer.retrieveProduct(createProduct(stageFile), destDir);
+ }
+
+ @VisibleForTesting
+ static DataTransfer createDataTransfer(PgeMetadata pgeMetadata,
+ Logger logger) throws InstantiationException {
+ if (pgeMetadata.getMetadata(QUERY_CLIENT_TRANSFER_SERVICE_FACTORY) !=
null) {
+ return GenericFileManagerObjectFactory
+ .getDataTransferServiceFromFactory(pgeMetadata
+ .getMetadata(QUERY_CLIENT_TRANSFER_SERVICE_FACTORY));
+ } else {
+ logger.log(Level.INFO, "Using default DataTransferer");
+ return new RemoteDataTransferFactory().createDataTransfer();
+ }
+ }
+
+ @VisibleForTesting
+ static void setFileManagerUrl(DataTransfer dataTransferer,
+ PgeMetadata pgeMetadata, Logger logger) throws MalformedURLException {
+ String filemgrUrl = pgeMetadata.getMetadata(QUERY_FILE_MANAGER_URL);
+ if (filemgrUrl != null) {
+ dataTransferer.setFileManagerUrl(new URL(filemgrUrl));
+ } else {
+ logger.log(Level.WARNING, "Metadata field [" + QUERY_FILE_MANAGER_URL
+ + "] was not set, if DataTranferer requires filemgr server,"
+ + " your transfers will fail");
+ }
+ }
+
+ @VisibleForTesting
+ static Product createProduct(URI path) {
+ Product product = new Product();
+ Reference reference = new Reference();
+ reference.setDataStoreReference(path.toString());
+ product.setProductReferences(Lists.newArrayList(reference));
+ return product;
+ }
+}
Propchange:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileManagerFileStager.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java?rev=1332872&view=auto
==============================================================================
---
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java
(added)
+++
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java
Tue May 1 22:36:17 2012
@@ -0,0 +1,123 @@
+/*
+ * 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.oodt.cas.pge.staging;
+
+//OODT static imports
+import static
org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys.QUERY_FILE_MANAGER_URL;
+
+//JDK imports
+import java.io.File;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+//OODT imports
+import org.apache.commons.lang.Validate;
+import org.apache.oodt.cas.filemgr.structs.Product;
+import org.apache.oodt.cas.filemgr.structs.Reference;
+import org.apache.oodt.cas.filemgr.structs.exceptions.CatalogException;
+import org.apache.oodt.cas.filemgr.system.XmlRpcFileManagerClient;
+import org.apache.oodt.cas.pge.config.FileStagingInfo;
+import org.apache.oodt.cas.pge.metadata.PgeMetadata;
+
+//Google imports
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+
+/**
+ * Responsible for transferring Product files to a directory accessible by
+ * by CAS-PGE.
+ *
+ * @author bfoster (Brian Foster)
+ */
+public abstract class FileStager {
+
+ public void stageFiles(FileStagingInfo fileStagingInfo,
+ PgeMetadata pgeMetadata, Logger logger) throws Exception {
+ logger.log(Level.INFO, "Creating staging directory ["
+ + fileStagingInfo.getStagingDir() + "]");
+ new File(fileStagingInfo.getStagingDir()).mkdirs();
+ for (String file : fileStagingInfo.getFilePaths()) {
+ File fileHandle = new File(file);
+ if (fileStagingInfo.isForceStaging() || !fileHandle.exists()) {
+ logger.log(Level.INFO, "Staging file [" + file
+ + "] to directory ["
+ + fileStagingInfo.getStagingDir() + "]");
+ stageFile(asURI(file), new File(fileStagingInfo.getStagingDir()),
+ pgeMetadata, logger);
+ }
+ }
+ if (!fileStagingInfo.getProductIds().isEmpty()) {
+ XmlRpcFileManagerClient fmClient =
createFileManagerClient(pgeMetadata);
+ for (String productId : fileStagingInfo.getProductIds()) {
+ logger.log(Level.INFO, "Staging product [" + productId
+ + "] to directory ["
+ + fileStagingInfo.getStagingDir() + "]");
+ for (URI uri : getProductReferences(productId, fmClient)) {
+ logger.log(Level.INFO, "Staging product [" + productId
+ + "] reference [" + uri
+ + "] to directory ["
+ + fileStagingInfo.getStagingDir() + "]");
+ stageFile(uri, new File(fileStagingInfo.getStagingDir()),
+ pgeMetadata, logger);
+ }
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static XmlRpcFileManagerClient createFileManagerClient(PgeMetadata
pgeMetadata)
+ throws Exception {
+ String filemgrUrl = pgeMetadata.getMetadata(QUERY_FILE_MANAGER_URL);
+ if (filemgrUrl == null) {
+ throw new Exception("Must specify [" + QUERY_FILE_MANAGER_URL
+ + "] if you want to stage product IDs");
+ }
+ return new XmlRpcFileManagerClient(new URL(filemgrUrl));
+ }
+
+ @VisibleForTesting
+ static List<URI> getProductReferences(
+ String productId, XmlRpcFileManagerClient fmClient)
+ throws URISyntaxException, CatalogException {
+ List<URI> files = Lists.newArrayList();
+ Product product = new Product();
+ product.setProductId(productId);
+ List<Reference> refs = fmClient.getProductReferences(product);
+ for (Reference ref : refs) {
+ files.add(new URI(ref.getDataStoreReference()));
+ }
+ return files;
+ }
+
+ @VisibleForTesting
+ static URI asURI(String path) throws URISyntaxException {
+ Validate.notNull(path, "path must not be null");
+
+ URI uri = URI.create(path);
+ if (uri.getScheme() == null) {
+ uri = URI.create("file://" + new File(path).getAbsolutePath());
+ }
+ return uri;
+ }
+
+ protected abstract void stageFile(URI stageFile, File destDir,
+ PgeMetadata pgeMetadata, Logger logger) throws Exception;
+}
Propchange:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/staging/FileStager.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java?rev=1332872&view=auto
==============================================================================
---
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java
(added)
+++
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java
Tue May 1 22:36:17 2012
@@ -0,0 +1,93 @@
+/*
+ * 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.oodt.cas.pge.util;
+
+//JDK imports
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+//OODT imports
+import org.apache.oodt.cas.pge.ConfigFilePropertyAdder;
+import org.apache.oodt.cas.pge.PGETaskInstance;
+import org.apache.oodt.cas.pge.config.PgeConfigBuilder;
+import org.apache.oodt.cas.pge.staging.FileStager;
+import org.apache.oodt.cas.pge.writers.SciPgeConfigFileWriter;
+
+/**
+ * Factory for creating {@link Object}s.
+ *
+ * @author bfoster (Brian Foster)
+ */
+public class GenericPgeObjectFactory {
+
+ private GenericPgeObjectFactory() {}
+
+ public static PGETaskInstance createPGETaskInstance(
+ String clazz, Logger logger) {
+ try {
+ return (PGETaskInstance) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ logger.log(Level.SEVERE, "Failed to create PGETaskInstance ["
+ + clazz + "] : " + e.getMessage(), e);
+ return null;
+ }
+ }
+
+ public static PgeConfigBuilder createPgeConfigBuilder(
+ String clazz, Logger logger) {
+ try {
+ return (PgeConfigBuilder) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ logger.log(Level.SEVERE, "Failed to create PgeConfigBuilder ["
+ + clazz + "] : " + e.getMessage(), e);
+ return null;
+ }
+ }
+
+ public static ConfigFilePropertyAdder createConfigFilePropertyAdder(
+ String clazz, Logger logger) {
+ try {
+ return (ConfigFilePropertyAdder) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ logger.log(Level.SEVERE, "Failed to create ConfigFilePropertyAdder ["
+ + clazz + "] : " + e.getMessage(), e);
+ return null;
+ }
+ }
+
+ public static FileStager createFileStager(
+ String clazz, Logger logger) {
+ try {
+ return (FileStager) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ logger.log(Level.SEVERE, "Failed to create FileStager ["
+ + clazz + "] : " + e.getMessage(), e);
+ return null;
+ }
+ }
+
+ public static SciPgeConfigFileWriter createSciPgeConfigFileWriter(
+ String clazz, Logger logger) {
+ try {
+ return (SciPgeConfigFileWriter) Class.forName(clazz).newInstance();
+ } catch (Exception e) {
+ logger.log(Level.SEVERE, "Failed to create SciPgeConfigFileWriter ["
+ + clazz + "] : " + e.getMessage(), e);
+ return null;
+ }
+ }
+}
Propchange:
oodt/trunk/pge/src/main/java/org/apache/oodt/cas/pge/util/GenericPgeObjectFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added:
oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java?rev=1332872&view=auto
==============================================================================
--- oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java
(added)
+++ oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java
Tue May 1 22:36:17 2012
@@ -0,0 +1,160 @@
+/*
+ * 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.oodt.cas.pge.staging;
+
+//OODT static imports
+import static
org.apache.oodt.cas.pge.staging.TestFileStager.ProductIdMatcher.eqProductId;
+
+//EasyMock static imports
+import static org.easymock.EasyMock.createStrictMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+
+//JDK imports
+import java.io.File;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.List;
+
+//OODT imports
+import org.apache.oodt.cas.filemgr.structs.Product;
+import org.apache.oodt.cas.filemgr.structs.Reference;
+import org.apache.oodt.cas.filemgr.structs.exceptions.CatalogException;
+import org.apache.oodt.cas.filemgr.system.XmlRpcFileManager;
+import org.apache.oodt.cas.filemgr.system.XmlRpcFileManagerClient;
+import org.apache.oodt.cas.pge.metadata.PgeMetadata;
+import org.apache.oodt.cas.pge.metadata.PgeTaskMetKeys;
+
+//EasyMock imports
+import org.easymock.EasyMock;
+import org.easymock.IArgumentMatcher;
+
+//Google imports
+import com.google.common.collect.Lists;
+
+//JUnit imports
+import junit.framework.TestCase;
+
+/**
+ * Test class for {@link FileStager}.
+ *
+ * @author bfoster (Brian Foster)
+ */
+public class TestFileStager extends TestCase {
+
+ public void testCreateFileManagerClient() throws Exception {
+ // Test throws case.
+ PgeMetadata pgeMetadata = createStrictMock(PgeMetadata.class);
+ expect(pgeMetadata.getMetadata(PgeTaskMetKeys.QUERY_FILE_MANAGER_URL))
+ .andReturn(null);
+ replay(pgeMetadata);
+
+ try {
+ FileStager.createFileManagerClient(pgeMetadata);
+ fail("Should have thrown Exception");
+ } catch (Exception e) { /* expect throw */
+ }
+ verify(pgeMetadata);
+
+ // Test success case.
+ System.setProperty("filemgr.catalog.factory", "");
+ System.setProperty("filemgr.repository.factory", "");
+ int port = 9876;
+ XmlRpcFileManager filemgr = new XmlRpcFileManager(port);
+ String filemgrUrl = "http://localhost:"; + port;
+ pgeMetadata = createStrictMock(PgeMetadata.class);
+ expect(pgeMetadata.getMetadata(PgeTaskMetKeys.QUERY_FILE_MANAGER_URL))
+ .andReturn(filemgrUrl);
+ replay(pgeMetadata);
+
+ assertEquals(filemgrUrl, FileStager.createFileManagerClient(pgeMetadata)
+ .getFileManagerUrl().toString());
+ verify(pgeMetadata);
+ filemgr.shutdown();
+ }
+
+ public void testGetProductReferences() throws CatalogException,
+ URISyntaxException {
+ String productId = "12345";
+
+ String uri1 = "file:///path/to/file1";
+ String uri2 = "file:///path/to/file2";
+ Reference ref1 = new Reference();
+ ref1.setDataStoreReference(uri1);
+ Reference ref2 = new Reference();
+ ref2.setDataStoreReference(uri2);
+
+ XmlRpcFileManagerClient fmClient =
createStrictMock(XmlRpcFileManagerClient.class);
+ expect(fmClient.getProductReferences(eqProductId(productId))).andReturn(
+ Lists.newArrayList(ref1, ref2));
+ replay(fmClient);
+
+ List<URI> uris = FileStager.getProductReferences(productId, fmClient);
+ assertEquals(2, uris.size());
+ assertTrue(uris.contains(new URI("file:///path/to/file1")));
+ assertTrue(uris.contains(new URI("file:///path/to/file2")));
+
+ verify(fmClient);
+ }
+
+ public void testAsURI() throws URISyntaxException {
+ String absoluteHttpUri = "http://somewhere.com/path/to/data.dat";;
+ String absoluteFileUri = "file:///path/to/data.dat";
+ String relativePath = "path/to/data.dat";
+ String absolutePath = "/path/to/data.dat";
+
+ assertEquals("http://somewhere.com/path/to/data.dat";,
+ FileStager.asURI(absoluteHttpUri).toString());
+ assertEquals("file:///path/to/data.dat", FileStager
+ .asURI(absoluteFileUri).toString());
+ assertEquals("file://" + new File("").getAbsolutePath()
+ + "/path/to/data.dat", FileStager.asURI(relativePath).toString());
+ assertEquals("file:///path/to/data.dat", FileStager.asURI(absolutePath)
+ .toString());
+ }
+
+ public static class ProductIdMatcher implements IArgumentMatcher {
+
+ private String productId;
+
+ public ProductIdMatcher(String productId) {
+ this.productId = productId;
+ }
+
+ @Override
+ public void appendTo(StringBuffer buffer) {
+ buffer.append("eqProduct(");
+ buffer.append(Product.class.getName());
+ buffer.append(" with product id [");
+ buffer.append(productId + "])");
+ }
+
+ @Override
+ public boolean matches(Object obj) {
+ if (obj instanceof Product) {
+ return productId.equals(((Product) obj).getProductId());
+ }
+ return false;
+ }
+
+ public static Product eqProductId(String productId) {
+ EasyMock.reportMatcher(new ProductIdMatcher(productId));
+ return null;
+ }
+ }
+}
Propchange:
oodt/trunk/pge/src/test/org/apache/oodt/cas/pge/staging/TestFileStager.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
