Support for benchmark

haas-imagej-client/pom.xml

@@ -85,7 +85,16 @@
 			<artifactId>haas-java-client</artifactId>
 			<version>0.0.1-SNAPSHOT</version>
 		</dependency>
-		<!-- https://mvnrepository.com/artifact/net.imagej/imagej-launcher -->
+		<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-jdk14 -->
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-jdk14</artifactId>
+		</dependency>
 
 	</dependencies>
 </project>

haas-imagej-client/src/main/java/cz/it4i/fiji/haas/BenchmarkJobManager.java

+package cz.it4i.fiji.haas;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.Collections;
+
+import cz.it4i.fiji.haas.JobManager.JobInfo;
+import cz.it4i.fiji.haas_java_client.JobState;
+import net.imagej.updater.util.Progress;
+
+public class BenchmarkJobManager {
+	private JobManager jobManager;
+	private Progress progress;
+	private Path workDirectory;
+	private static final String CONFIG_FOR_MODIFICATION = "not-set-config.yaml";
+	private static final String CONFIG_MODIFIED = "config.yaml";
+	
+	public BenchmarkJobManager(Path workDirectory, Progress progress) throws IOException {
+		this.workDirectory = workDirectory;
+		jobManager = new JobManager(workDirectory, TestingConstants.getSettings(3, 6));
+		this.progress = progress;
+	}
+	
+	public JobInfo startJob() throws IOException {
+		JobInfo jobInfo = jobManager.startJob( Collections.emptyList(), null);
+		jobInfo.waitForStart();
+		if(jobInfo.getState() != JobState.Running) {
+			throw new IllegalStateException("start of job: " + jobInfo + " failed");
+		}
+		ByteArrayOutputStream os = new ByteArrayOutputStream();
+		jobInfo.downloadFileData(CONFIG_FOR_MODIFICATION,os);
+		byte[]data  = updateConfigFile(os.toByteArray());
+		
+		jobInfo.uploadFile (new ByteArrayInputStream(data),CONFIG_MODIFIED, data.length ,Instant.now().getEpochSecond());
+		return jobInfo;
+	}
+
+	private byte[] updateConfigFile(byte[] data) throws IOException {
+		return data;
+		
+	}
+
+	
+}

haas-imagej-client/src/main/java/cz/it4i/fiji/haas/Job.java

@@ -24,7 +24,6 @@ import net.imagej.updater.util.Progress;
 
 public class Job {
 
-	
 	private static final String JOB_HAS_DATA_TO_DOWNLOAD_PROPERTY = "job.needDownload";
 
 	public static boolean isJobPath(Path p) {
@@ -46,37 +45,39 @@ public class Job {
 	private Long jobId;
 
 	final private Progress dummy = new Progress() {
-		
+
 		@Override
 		public void setTitle(String title) {
 		}
-		
+
 		@Override
 		public void setItemCount(int count, int total) {
 		}
-		
+
 		@Override
 		public void setCount(int count, int total) {
 		}
-		
+
 		@Override
 		public void itemDone(Object item) {
 		}
-		
+
 		@Override
 		public void done() {
 		}
-		
+
 		@Override
 		public void addItem(Object item) {
 		}
 	};
-	
-	public Job(Path path, Collection<Path> files, Supplier<HaaSClient> haasClientSupplier, Progress progress) throws IOException {
+
+	public Job(Path basePath, Collection<Path> files, Supplier<HaaSClient> haasClientSupplier, Progress progress)
+			throws IOException {
 		this(haasClientSupplier);
 		HaaSClient client = this.haasClientSupplier.get();
-		long id = client.start(files, "TestOutRedirect", Collections.emptyList(), new P_ProgressNotifierAdapter(progress));
-		jobDir = path.resolve("" + id);
+		long id = client.start(files, "TestOutRedirect", Collections.emptyList(),
+				new P_ProgressNotifierAdapter(progress));
+		jobDir = basePath.resolve("" + id);
 		Files.createDirectory(jobDir);
 		updateState();
 	}
@@ -132,6 +133,11 @@ public class Job {
 		}
 	}
 
+	public void downloadFileData(String fileName, OutputStream bos) {
+		haasClientSupplier.get().downloadFileData(jobId,fileName, bos);
+	}
+	
+
 	public JobState getState() {
 		return state;
 	}
@@ -182,8 +188,6 @@ public class Job {
 	private static long getJobId(Path path) {
 		return Long.parseLong(path.getFileName().toString());
 	}
-	
-	
 
 	private class P_ProgressNotifierAdapter implements ProgressNotifier {
 		private Progress progress;
@@ -216,7 +220,9 @@ public class Job {
 		public void done() {
 			progress.done();
 		}
-		
-		
+
 	}
+
+	
+
 }

haas-imagej-client/src/main/java/cz/it4i/fiji/haas/JobManager.java

 package cz.it4i.fiji.haas;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Calendar;
@@ -9,6 +11,9 @@ import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.stream.Collectors;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import cz.it4i.fiji.haas_java_client.HaaSClient;
 import cz.it4i.fiji.haas_java_client.JobState;
 import cz.it4i.fiji.haas_java_client.Settings;
@@ -17,6 +22,8 @@ import net.imagej.updater.util.Progress;
 
 public class JobManager {
 
+	private static Logger log = LoggerFactory.getLogger(cz.it4i.fiji.haas.JobManager.class);
+
 	private Path workDirectory;
 
 	private Collection<Job> jobs = new LinkedList<>();
@@ -25,7 +32,6 @@ public class JobManager {
 
 	private Settings settings;
 
-	
 	public JobManager(Path workDirectory, Settings settings) throws IOException {
 		this.workDirectory = workDirectory;
 		this.settings = settings;
@@ -39,8 +45,20 @@ public class JobManager {
 
 	}
 
-	public void startJob(Path path, Collection<Path> files, Progress progress) throws IOException {
-		jobs.add(new Job(path, files, this::getHaasClient, progress));
+	public JobInfo startJob(Collection<Path> files, Progress progress) throws IOException {
+		Job job;
+		jobs.add(job = new Job(workDirectory, files, this::getHaasClient, progress));
+		return new JobInfo(job) {
+			@Override
+			public JobState getState() {
+				try {
+					job.updateState();
+				} catch (IOException e) {
+					log.error(e.getMessage(), e);
+				}
+				return super.getState();
+			}
+		};
 	}
 
 	public Iterable<JobInfo> getJobsNeedingDownload() {
@@ -58,6 +76,12 @@ public class JobManager {
 
 	}
 
+	public JobState getState(long id) {
+		return getHaasClient().obtainJobInfo(id).getState();
+	}
+	
+	
+
 	private HaaSClient getHaasClient() {
 		if (haasClient == null) {
 			haasClient = new HaaSClient(settings);
@@ -65,7 +89,6 @@ public class JobManager {
 		return haasClient;
 	}
 
-
 	public static class JobInfo extends ObservableValueBase<JobInfo> {
 
 		private Job job;
@@ -102,7 +125,13 @@ public class JobManager {
 			job.download(progress);
 			fireValueChangedEvent();
 		}
+		
+		public void waitForStart() {
+			// TODO Auto-generated method stub
+			
+		}
 
+		
 		public void updateInfo() throws IOException {
 			job.updateState();
 		}
@@ -111,10 +140,25 @@ public class JobManager {
 		public JobInfo getValue() {
 			return this;
 		}
+		
+		public void downloadFileData(String fileName, OutputStream bos) {
+			job.downloadFileData(fileName, bos);
+		}
+
+		public void uploadFile(ByteArrayInputStream byteArrayInputStream, String configModified, int length,
+				long epochSecond) {
+			
+			//TODO
+		}
 
 		private String getStringFromTimeSafely(Calendar time) {
 			return time != null ? time.getTime().toString() : "N/A";
 		}
 
+
+
+
+
 	}
+
 }

haas-imagej-client/src/main/java/cz/it4i/fiji/haas/RunWithHaaS.java

@@ -49,9 +49,8 @@ public class RunWithHaaS implements Command {
 	@Override
 	public void run() {
 		try {
-			jobManager = new JobManager(getWorkingDirectoryPath(),TestingConstants.getSettings());
-			jobManager.startJob(getWorkingDirectoryPath(), getContent(dataDirectory),
-					ModalDialogs.doModal(new ProgressDialog(getFrame())));
+			jobManager = new JobManager(getWorkingDirectoryPath(), TestingConstants.getSettings());
+			jobManager.startJob(getContent(dataDirectory), ModalDialogs.doModal(new ProgressDialog(getFrame())));
 		} catch (IOException e) {
 			log.error(e);
 		}

haas-java-client/src/main/java/cz/it4i/fiji/haas_java_client/HaaSClient.java

 package cz.it4i.fiji.haas_java_client;
 
 import java.io.IOException;
+import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -14,7 +15,9 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.xml.rpc.ServiceException;
 
@@ -210,7 +213,7 @@ public class HaaSClient {
 				public java.util.Calendar getEndTime() {
 					return info.getEndTime();
 				};
-				
+
 				public Calendar getCreationTime() {
 					return info.getCreationTime();
 				};
@@ -235,20 +238,25 @@ public class HaaSClient {
 	}
 
 	public void download(long jobId, Path workDirectory, final ProgressNotifier notifier) {
+		download(jobId, workDirectory, notifier, val -> true);
+	}
+
+	public void download(long jobId, Path workDirectory, final ProgressNotifier notifier, Predicate<String> function) {
 		try {
 			notifier.setTitle("Downloading");
 			FileTransferMethodExt ft = getFileTransfer().getFileTransferMethod(jobId, getSessionID());
 			try (ScpClient scpClient = getScpClient(ft)) {
-				String[] files = getFileTransfer().listChangedFilesForJob(jobId, getSessionID());
+				String[] filesArray = getFileTransfer().listChangedFilesForJob(jobId, getSessionID());
+				Stream<String> files = Arrays.asList(filesArray).stream().filter(function);
 				List<Long> fileSizes = getSizes(
-						Arrays.asList(files).stream()
-								.map(filename -> "'" + ft.getSharedBasepath() + "/" + filename + "'")
-								.collect(Collectors.toList()),
+						files.map(filename -> "'" + ft.getSharedBasepath() + "/" + filename + "'").collect(
+								Collectors.toList()),
 						scpClient, new P_ProgressNotifierDecorator4Size(notifier));
 				final long totalFileSize = fileSizes.stream().mapToLong(i -> i.longValue()).sum();
-				TransferFileProgressForHaaSClient progress =new TransferFileProgressForHaaSClient(totalFileSize, notifier);
+				TransferFileProgressForHaaSClient progress = new TransferFileProgressForHaaSClient(totalFileSize,
+						notifier);
 				int idx = 0;
-				for (String fileName : files) {
+				for (String fileName : (Iterable<String>) files::iterator) {
 					fileName = fileName.replaceFirst("/", "");
 					Path rFile = workDirectory.resolve(fileName);
 					if (!Files.exists(rFile.getParent())) {
@@ -271,6 +279,17 @@ public class HaaSClient {
 		}
 	}
 
+	public void downloadFileData(long jobId, String fileName, OutputStream os) {
+		try {
+			FileTransferMethodExt ft = getFileTransfer().getFileTransferMethod(jobId, getSessionID());
+			try (ScpClient scpClient = getScpClient(ft)) {
+				scpClient.download(fileName, os, new TransferFileProgressForHaaSClient(0, dummyNotifier));
+			}
+		} catch (IOException | JSchException | ServiceException e) {
+			throw new HaaSClientException(e);
+		}
+	}
+
 	private List<Long> getSizes(List<String> asList, ScpClient scpClient, ProgressNotifier notifier)
 			throws JSchException, IOException {
 		List<Long> result = new LinkedList<>();
@@ -428,4 +447,5 @@ public class HaaSClient {
 			notifier.done();
 		}
 	}
+
 }

java-scpclient/src/main/java/cz/it4i/fiji/scpclient/ScpClient.java

@@ -66,6 +66,13 @@ public class ScpClient implements Closeable {
 	}
 
 	public boolean download(String lfile, Path rfile, TransferFileProgress progress) throws JSchException, IOException {
+		try(OutputStream os = Files.newOutputStream(rfile)) {
+			return download(lfile, os, progress);
+		}
+	}
+
+	public boolean download(String lfile, OutputStream os, TransferFileProgress progress)
+			throws JSchException, IOException {
 		Session session = connectionSession();
 
 		// exec 'scp -f rfile' remotely
@@ -125,25 +132,24 @@ public class ScpClient implements Closeable {
 					out.flush();
 
 					// read a content of lfile
-					try (OutputStream fos = Files.newOutputStream(rfile)) {
-						int foo;
-						while (true) {
-							if (buf.length < filesize)
-								foo = buf.length;
-							else
-								foo = (int) filesize;
-							foo = in.read(buf, 0, foo);
-							if (foo < 0) {
-								// error
-								break;
-							}
-							fos.write(buf, 0, foo);
-							progress.dataTransfered(foo);
-							filesize -= foo;
-							if (filesize == 0L)
-								break;
+					int foo;
+					while (true) {
+						if (buf.length < filesize)
+							foo = buf.length;
+						else
+							foo = (int) filesize;
+						foo = in.read(buf, 0, foo);
+						if (foo < 0) {
+							// error
+							break;
 						}
+						os.write(buf, 0, foo);
+						progress.dataTransfered(foo);
+						filesize -= foo;
+						if (filesize == 0L)
+							break;
 					}
+
 					if (checkAck(in) != 0) {
 						return false;
 					}
@@ -167,11 +173,16 @@ public class ScpClient implements Closeable {
 	}
 
 	public boolean upload(Path file, String rfile, TransferFileProgress progress) throws JSchException, IOException {
+		try (InputStream is = Files.newInputStream(file)) {
+			return upload(is, file.getFileName().toString(), file.toFile().length(), file.toFile().lastModified(),
+					rfile, progress);
+		}
+	}
 
+	public boolean upload(InputStream is, String fileName, long length, long lastModified, String rfile,
+			TransferFileProgress progress) throws JSchException, IOException {
 		Session session = connectionSession();
-
 		boolean ptimestamp = true;
-
 		// exec 'scp -t rfile' remotely
 		String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + rfile;
 		Channel channel = session.openChannel("exec");
@@ -184,10 +195,10 @@ public class ScpClient implements Closeable {
 			}
 
 			if (ptimestamp) {
-				command = "T " + (file.toFile().lastModified() / 1000) + " 0";
+				command = "T " + (lastModified / 1000) + " 0";
 				// The access time should be sent here,
 				// but it is not accessible with JavaAPI ;-<
-				command += (" " + (file.toFile().lastModified() / 1000) + " 0\n");
+				command += (" " + (lastModified / 1000) + " 0\n");
 				out.write(command.getBytes());
 				out.flush();
 				if (checkAck(in) != 0) {
@@ -196,9 +207,9 @@ public class ScpClient implements Closeable {
 			}
 
 			// send "C0644 filesize filename", where filename should not include '/'
-			long filesize = file.toFile().length();
+			long filesize = length;
 			command = "C0644 " + filesize + " ";
-			command += file.getFileName().toString();
+			command += fileName;
 			command += "\n";
 			out.write(command.getBytes());
 			out.flush();
@@ -207,14 +218,12 @@ public class ScpClient implements Closeable {
 			}
 			byte[] buf = new byte[getBufferSize()];
 			// send a content of lfile
-			try (InputStream fis = Files.newInputStream(file)) {
-				while (true) {
-					int len = fis.read(buf, 0, buf.length);
-					if (len <= 0)
-						break;
-					out.write(buf, 0, len); // out.flush();
-					progress.dataTransfered(len);
-				}
+			while (true) {
+				int len = is.read(buf, 0, buf.length);
+				if (len <= 0)
+					break;
+				out.write(buf, 0, len); // out.flush();
+				progress.dataTransfered(len);
 			}
 			// send '\0'
 			buf[0] = 0;
@@ -233,35 +242,35 @@ public class ScpClient implements Closeable {
 
 	public long size(String lfile) throws JSchException, IOException {
 		Session session = connectionSession();
-	
+
 		// exec 'scp -f rfile' remotely
 		String command = "scp -f " + lfile;
 		Channel channel = session.openChannel("exec");
-	
+
 		try {
 			((ChannelExec) channel).setCommand(command);
-	
+
 			// get I/O streams for remote scp
 			try (OutputStream out = channel.getOutputStream(); InputStream in = channel.getInputStream()) {
-	
+
 				channel.connect();
-	
+
 				byte[] buf = new byte[getBufferSize()];
-	
+
 				// send '\0'
 				buf[0] = 0;
 				out.write(buf, 0, 1);
 				out.flush();
-	
+
 				while (true) {
 					int c = checkAck(in);
 					if (c != 'C') {
 						break;
 					}
-	
+
 					// read '0644 '
 					in.read(buf, 0, 5);
-	
+
 					long filesize = 0L;
 					while (true) {
 						if (in.read(buf, 0, 1) < 0) {
@@ -273,10 +282,10 @@ public class ScpClient implements Closeable {
 						filesize = filesize * 10L + (long) (buf[0] - '0');
 					}
 					return filesize;
-	
+
 				}
 			}
-	
+
 		} finally {
 			channel.disconnect();
 		}