#25 Use URIs and own FileOperations implementations for HDFS

eu.mondo.sam.core/pom.xml

@@ -66,7 +66,19 @@
 			<artifactId>jackson-mapper-asl</artifactId>
 			<version>1.6.0</version>
 		</dependency>
-
+
+		<dependency>
+			<groupId>commons-io</groupId>
+			<artifactId>commons-io</artifactId>
+			<version>2.0.1</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.httpcomponents</groupId>
+			<artifactId>fluent-hc</artifactId>
+			<version>4.5.1</version>
+		</dependency>
+
 	</dependencies>
 
 </project>

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/BenchmarkEngine.java

@@ -1,7 +1,7 @@
 package eu.mondo.sam.core;
 
 import java.io.IOException;
-import java.nio.file.Path;
+import java.net.URI;
 
 import eu.mondo.sam.core.phases.AtomicPhase;
 import eu.mondo.sam.core.results.BenchmarkResult;
@@ -33,7 +33,7 @@ public class BenchmarkEngine {
 	/**
 	 * Instantiates the benchmarkResult variable.
 	 */
-	public BenchmarkEngine(Path resultsPath) {
+	public BenchmarkEngine(URI resultsPath) {
 		benchmarkResult = new BenchmarkResult(resultsPath);
 	}
 

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/results/BenchmarkResult.java

@@ -1,12 +1,14 @@
 package eu.mondo.sam.core.results;
 
 import java.io.IOException;
-import java.nio.file.Path;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 
 import org.codehaus.jackson.annotate.JsonProperty;
 
+import eu.mondo.sam.core.util.UriUtils;
+
 /**
  * Contains the results of the entire benchmark process, as consists of
  * PhaseResult objects as a list. Also responsible for using the appropriate
@@ -32,7 +34,7 @@ public class BenchmarkResult {
 	@JsonProperty("PhaseResults")
 	private List<PhaseResult> phaseResults;
 
-	private final Path resultsPath;
+	private final URI resultsPath;
 
 	/**
 	 * Includes of ResultSerializer instances. The elements in the list are
@@ -45,7 +47,7 @@ public class BenchmarkResult {
 	 * Instantiates the phaseResults list and the serializers as well.
 	 * @param resultsPath the directory where the results should be output 
 	 */
-	public BenchmarkResult(Path resultsPath) {
+	public BenchmarkResult(URI resultsPath) {
 		this.resultsPath = resultsPath;
 		phaseResults = new ArrayList<PhaseResult>();
 		serializers = new ArrayList<ResultSerializer>();
@@ -97,7 +99,7 @@ public void publishResults() throws IOException {
 		int runIndex = caseDescriptor.getRunIndex();
 		String fileName = tool + "-" + benchCase + "-" + scenario
 				+ "-Size" + size + "-Index" + runIndex;
-		Path resultFilePath = resultsPath.resolve(fileName);
+		URI resultFilePath = UriUtils.appendSegment(resultsPath, fileName);
 
 		for (ResultSerializer serializer : serializers) {
 			serializer.serialize(this, resultFilePath);

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/results/JsonSerializer.java

@@ -1,17 +1,19 @@
 package eu.mondo.sam.core.results;
 
+import java.io.File;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
-import java.nio.file.StandardOpenOption;
+import java.net.URI;
 
+import org.apache.commons.io.FileUtils;
 import org.codehaus.jackson.JsonGenerationException;
 import org.codehaus.jackson.map.JsonMappingException;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.SerializationConfig;
 
+import eu.mondo.sam.core.util.FileOperations;
+import eu.mondo.sam.core.util.UriUtils;
+
 /**
  * Responsible for publishing the results of benchmarking.
  * 
@@ -36,7 +38,7 @@ public class JsonSerializer implements ResultSerializer {
 	 * @throws IOException
 	 *                 if some error occur during the JSON serialization
 	 */
-	public void serialize(BenchmarkResult benchmarkResult, Path resultFilePathWithoutExtension)
+	public void serialize(BenchmarkResult benchmarkResult, URI resultFilePathWithoutExtension)
 			throws IOException {
 		ObjectMapper mapper = new ObjectMapper();
 		// to enable standard indentation ("pretty-printing"):
@@ -54,13 +56,12 @@ public void serialize(BenchmarkResult benchmarkResult, Path resultFilePathWithou
 				false);
 
 		try {
-			String fileNameWithExtension = resultFilePathWithoutExtension.getFileName().toString() + ".json";
-			Path resultFilePath = resultFilePathWithoutExtension.resolveSibling(fileNameWithExtension);
-			Path tempFilePath = Files.createTempFile(null, null);
-			OutputStream outputStream = Files.newOutputStream(tempFilePath, StandardOpenOption.CREATE);
+			URI resultFilePath = UriUtils.appendSuffix(resultFilePathWithoutExtension, ".json");
+			File tempFile = File.createTempFile("mondo-sam", null);
+			OutputStream outputStream = FileUtils.openOutputStream(tempFile);
 			mapper.writeValue(outputStream, benchmarkResult);
 			outputStream.close();
-			Files.copy(tempFilePath, resultFilePath, StandardCopyOption.REPLACE_EXISTING);
+			FileOperations.copy(tempFile, resultFilePath);
 		} catch (JsonGenerationException e) {
 			throw new IOException(e);
 		} catch (JsonMappingException e) {

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/results/ResultSerializer.java

@@ -1,10 +1,10 @@
 package eu.mondo.sam.core.results;
 
 import java.io.IOException;
-import java.nio.file.Path;
+import java.net.URI;
 
 public interface ResultSerializer {
 
-	public void serialize(BenchmarkResult benchmarkResult, Path resultFilePathWithoutExtension)
+	public void serialize(BenchmarkResult benchmarkResult, URI resultFilePathWithoutExtension)
 			throws IOException;
 }

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/util/FileOperations.java

@@ -0,0 +1,57 @@
+package eu.mondo.sam.core.util;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.http.client.fluent.Request;
+import org.apache.http.client.fluent.Response;
+import org.apache.http.entity.ContentType;
+
+import com.google.common.base.Throwables;
+
+public class FileOperations {
+
+	private static final String WEBHDFS_SCHEME = "webhdfs";
+
+	private static URI makeWebHdfsUri(URI destination, String operation) {
+		try {
+			return new URI("http", "", destination.getHost(), destination.getPort(), "/webhdfs/v1/" + destination.getPath(), "op=" + operation, "");
+		} catch (URISyntaxException e) {
+			throw Throwables.propagate(e);
+		}
+	}
+
+	private static boolean isHdfs(URI destination) {
+		return (destination.getScheme() != null) && destination.getScheme().equals(WEBHDFS_SCHEME);
+	}
+
+	public static void copy(File source, URI destination) throws IOException {
+		if (isHdfs(destination)) {
+			URI endpointUri = makeWebHdfsUri(destination, "CREATE");
+			Response response = Request.Put(endpointUri).execute();
+			String location = response.returnResponse().getFirstHeader("Location").getValue();
+			Request.Put(location).bodyFile(source, ContentType.DEFAULT_BINARY).execute();
+		} else {
+			FileUtils.copyFile(source, new File(destination.getPath()));
+		}
+	}
+
+	public static void copy(URI source, File destination) throws IOException {
+		if (isHdfs(source)) {
+			Request.Get(makeWebHdfsUri(source, "OPEN")).execute().saveContent(destination);
+		} else {
+			FileUtils.copyFile(new File(source.getPath()), destination);
+		}
+	}
+
+	public static File loadModel(URI modelUriWithoutExtension, String postfix) throws IOException {
+		URI modelUriWithExtension = UriUtils.appendSuffix(modelUriWithoutExtension, postfix);
+		File tempFile = File.createTempFile("benchmark", null);
+		copy(modelUriWithExtension, tempFile);
+		return tempFile;
+	}
+
+}

eu.mondo.sam.core/src/main/java/eu/mondo/sam/core/util/UriUtils.java

@@ -0,0 +1,20 @@
+package eu.mondo.sam.core.util;
+
+import java.net.URI;
+
+import org.apache.commons.io.FilenameUtils;
+
+public class UriUtils {
+
+	public static URI appendSuffix(URI resultFilePathWithoutExtension, String suffix) {
+		String fileNameWithExtension = FilenameUtils.getName(resultFilePathWithoutExtension.toString()) + suffix;
+		URI resultFilePath = resultFilePathWithoutExtension.resolve(fileNameWithExtension);
+		return resultFilePath;
+	}
+
+	public static URI appendSegment(URI base, String segment) {
+		String name = FilenameUtils.getName(base.getPath());
+		return base.resolve(name + "/" + segment);
+	}
+
+}

eu.mondo.sam.test/src/test/java/eu/mondo/sam/test/WorkflowTest.java

@@ -42,7 +42,7 @@ public static void init() throws IllegalArgumentException, IOException {
 
 		token = new TestDataToken();
 
-		engine = new BenchmarkEngine(new File(".").toPath());
+		engine = new BenchmarkEngine(new File(".").toURI());
 		BenchmarkResult.removeAllSerializers();
 	}