feat: limit download after job submit

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

@@ -29,7 +29,11 @@ import cz.it4i.fiji.haas_java_client.JobState;
 import cz.it4i.fiji.haas_java_client.ProgressNotifier;
 import cz.it4i.fiji.haas_java_client.TransferFileProgressForHaaSClient;
 import net.imagej.updater.util.Progress;
-
+/***
+ * TASK - napojit na UI 
+ * @author koz01
+ *
+ */
 public class Job {
 
 	private static final String JOB_NAME = "job.name";
@@ -38,6 +42,10 @@ public class Job {
 
 	private static final String JOB_INFO_FILENAME = ".jobinfo";
 
+	private static final String JOB_NEEDS_DOWNLOAD = "job.needs_download";
+	
+	private static final String JOB_CAN_BE_DOWNLOADED = "job.needs_download";
+
 	public static boolean isJobPath(Path p) {
 		return isValidPath(p);
 	}
@@ -73,8 +81,10 @@ public class Job {
 		setJobDirectory(jobDirectory);
 		propertyHolder = new PropertyHolder(jobDir.resolve(JOB_INFO_FILENAME));
 		resumeUpload();
+		resumeDownload();
 	}
 
+	
 	private Job(JobManager4Job jobManager, Supplier<HaaSClient> haasClientSupplier) {
 		this.haasClientSupplier = haasClientSupplier;
 		this.jobManager = jobManager;
@@ -101,15 +111,26 @@ public class Job {
 	}
 
 	public void startDownload(Predicate<String> predicate, Progress notifier) throws IOException {
+		setProperty(JOB_NEEDS_DOWNLOAD, true);
 		Collection<String> files = getHaaSClient().getChangedFiles(jobId).stream().filter(predicate)
 				.collect(Collectors.toList());
 		synchronization.startDownload(files);
 	}
-
-	public boolean isUploading() {
-		return Boolean.parseBoolean(getProperty(JOB_NEEDS_UPLOAD));
+	
+	public void stopDownloadData() {
+		setProperty(JOB_NEEDS_DOWNLOAD, false);
+		try {
+			this.synchronization.stopUpload();
+		} catch (IOException e) {
+			log.error(e.getMessage(), e);
+			throw new RuntimeException(e);
+		}
 	}
-
+	
+	public boolean canBeDownload() {
+		return Boolean.parseBoolean(getProperty(JOB_CAN_BE_DOWNLOADED));
+	}
+	
 	public void uploadFile(String file, ProgressNotifier notifier) {
 		uploadFiles(Arrays.asList(file), notifier);
 	}
@@ -144,8 +165,12 @@ public class Job {
 	public void submit() {
 		HaaSClient client = getHaaSClient();
 		client.submitJob(jobId);
+		stopDownloadData();
+		setCanBeDownloaded(true);
 	}
 
+	
+
 	synchronized public long getId() {
 		if (jobId == null) {
 			jobId = getJobId(jobDir);
@@ -275,6 +300,9 @@ public class Job {
 					() -> haasClientSupplier.get().startFileTransfer(getId(), HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS),
 					jobDir, Executors.newFixedThreadPool(2), () -> {
 						setProperty(JOB_NEEDS_UPLOAD, false);
+					}, () -> {
+						setProperty(JOB_NEEDS_DOWNLOAD, false);
+						setCanBeDownloaded(false);
 					});
 		} catch (IOException e) {
 			log.error(e.getMessage(), e);
@@ -287,6 +315,13 @@ public class Job {
 			synchronization.resumeUpload();
 		}
 	}
+	
+	private synchronized void resumeDownload() {
+		if (Boolean.parseBoolean(getProperty(JOB_NEEDS_DOWNLOAD))) {
+			synchronization.resumeDownload();
+		}
+	}
+
 
 	private void setName(String name) {
 		setProperty(JOB_NAME, name);
@@ -320,5 +355,9 @@ public class Job {
 	private static long getJobId(Path path) {
 		return Long.parseLong(path.getFileName().toString());
 	}
+	
+	private void setCanBeDownloaded(boolean b) {
+		setProperty(JOB_CAN_BE_DOWNLOADED, b);
+	}
 
 }

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

@@ -43,13 +43,13 @@ public class Synchronization {
 	
 	
 	public Synchronization(Supplier<HaaSFileTransfer> fileTransferSupplier, Path workingDirectory,
-			ExecutorService service, Runnable uploadFinishedNotifier) throws IOException {
+			ExecutorService service, Runnable uploadFinishedNotifier, Runnable downloadFinishedNotifier) throws IOException {
 
 		this.workingDirectory = workingDirectory;
 		this.filesDownloaded = new PersistentIndex<>(workingDirectory.resolve(FILE_INDEX_DOWNLOADED_FILENAME),
 				pathResolver);
 		this.uploadProcess = createUploadProcess(fileTransferSupplier, service, uploadFinishedNotifier);
-		this.downloadProcess = createDownloadProcess(fileTransferSupplier, service, uploadFinishedNotifier);
+		this.downloadProcess = createDownloadProcess(fileTransferSupplier, service, downloadFinishedNotifier);
 	}
 
 	public synchronized void startUpload() throws IOException {
@@ -130,13 +130,13 @@ public class Synchronization {
 
 		@Override
 		protected void processItem(HaaSFileTransfer tr, String file) {
-			tr.download(file, workingDirectory);
 			filesDownloaded.insert(workingDirectory.resolve(file));
 			try {
 				filesDownloaded.storeToFile();
 			} catch (IOException e) {
 				log.error(e.getMessage(), e);
 			}
+			tr.download(file, workingDirectory);
 		}
 		
 	}

haas-spim-benchmark/src/main/java/cz/it4i/fiji/haas_spim_benchmark/core/BenchmarkJobManager.java

@@ -49,8 +49,7 @@ import net.imagej.updater.util.Progress;
 
 public class BenchmarkJobManager {
 
-	private static final String JOB_HAS_DATA_TO_DOWNLOAD_PROPERTY = "job.needDownload";
-
+	
 	private static Logger log = LoggerFactory
 			.getLogger(cz.it4i.fiji.haas_spim_benchmark.core.BenchmarkJobManager.class);
 
@@ -82,7 +81,7 @@ public class BenchmarkJobManager {
 			running = null;
 			job.submit();
 			job.setProperty(SPIM_OUTPUT_FILENAME_PATTERN, outputName);
-			setDownloaded(false);
+		
 		}
 
 		public JobState getState() {
@@ -97,10 +96,7 @@ public class BenchmarkJobManager {
 			job.stopUploadData();
 		}
 
-		public boolean isUploading() {
-			return job.isUploading();
-		}
-
+		
 		public synchronized CompletableFuture<JobState> getStateAsync(Executor executor) {
 			if (running != null) {
 				return running;
@@ -112,15 +108,17 @@ public class BenchmarkJobManager {
 			return result;
 		}
 
-		public void downloadData(Progress progress) throws IOException {
+		public void startDownload(Progress progress) throws IOException {
 			if (job.getState() == JobState.Finished) {
 				String filePattern = job.getProperty(SPIM_OUTPUT_FILENAME_PATTERN);
-				job.download(downloadFinishedData(filePattern), new P_ProgressNotifierAdapter(progress));
+				job.startDownload(downloadFinishedData(filePattern) , progress);
 			} else if (job.getState() == JobState.Failed || job.getState() == JobState.Canceled) {
-				job.download(downloadFailedData(), new P_ProgressNotifierAdapter(progress));
+				job.startDownload(downloadFailedData(), progress);
 			}
-
-			setDownloaded(true);
+		}
+		
+		public boolean canBeDownloaded() {
+			return job.canBeDownload();
 		}
 
 		public void downloadStatistics(Progress progress) throws IOException {
@@ -159,10 +157,6 @@ public class BenchmarkJobManager {
 			return false;
 		}
 
-		public boolean downloaded() {
-			return getDownloaded();
-		}
-
 		public void update() {
 			job.updateInfo();
 		}
@@ -407,15 +401,7 @@ public class BenchmarkJobManager {
 			return Stream.concat(nonTaskSpecificErrors.stream(), taskSpecificErrors).collect(Collectors.toList());
 		}
 
-		private void setDownloaded(boolean b) {
-			job.setProperty(JOB_HAS_DATA_TO_DOWNLOAD_PROPERTY, b + "");
-		}
-
-		private boolean getDownloaded() {
-			String downloadedStr = job.getProperty(JOB_HAS_DATA_TO_DOWNLOAD_PROPERTY);
-			return downloadedStr != null && Boolean.parseBoolean(downloadedStr);
-		}
-
+		
 	}
 
 	public BenchmarkJobManager(BenchmarkSPIMParameters params) throws IOException {

haas-spim-benchmark/src/main/java/cz/it4i/fiji/haas_spim_benchmark/ui/BenchmarkSPIMController.java

@@ -112,10 +112,10 @@ public class BenchmarkSPIMController extends BorderPane implements CloseableCont
 				job -> JavaFXRoutines.notNullValue(job, j -> !EnumSet.of(JobState.Running).contains(j.getState())));
 		
 		menu.addItem("Download result",
-				job -> executeWSCallAsync("Downloading data", p -> job.getValue().downloadData(p)),
+				job -> executeWSCallAsync("Downloading data", p -> job.getValue().startDownload(p)),
 				job -> JavaFXRoutines.notNullValue(job,
 						j -> EnumSet.of(JobState.Failed, JobState.Finished, JobState.Canceled).contains(j.getState())
-								&& !j.downloaded()));
+								&& !j.canBeDownloaded()));
 		menu.addItem("Download statistics",
 				job -> executeWSCallAsync("Downloading data", p -> job.getValue().downloadStatistics(p)),
 				job -> JavaFXRoutines.notNullValue(job, j -> j.getState() == JobState.Finished));

haas-spim-benchmark/src/test/java/cz/it4i/fiji/haas/RunBenchmark.java

@@ -34,7 +34,7 @@ public class RunBenchmark {
 				if (state == JobState.Configuring) {
 					job.startJob(new DummyProgress());
 				} else if (state != JobState.Running && state != JobState.Queued) {
-					job.downloadData(new DummyProgress());
+					job.startDownload(new DummyProgress());
 				} else if (state == JobState.Running) {
 					log.info(job.getSnakemakeOutput());
 				}