feat: support stop/start upload during file transfer

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

@@ -2,6 +2,7 @@ package cz.it4i.fiji.haas;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
@@ -124,7 +125,7 @@ public class Job {
 	public void stopDownloadData() {
 		setProperty(JOB_NEEDS_DOWNLOAD, false);
 		try {
-			this.synchronization.stopUpload();
+			this.synchronization.stopDownload();
 		} catch (IOException e) {
 			log.error(e.getMessage(), e);
 			throw new RuntimeException(e);
@@ -132,13 +133,15 @@ public class Job {
 	}
 	
 	public synchronized void resumeUpload() {
-		if (Boolean.parseBoolean(getProperty(JOB_NEEDS_UPLOAD))) {
+		if (needsUpload()) {
 			synchronization.resumeUpload();
 		}
 	}
 	
+	
+
 	public synchronized void resumeDownload() {
-		if (Boolean.parseBoolean(getProperty(JOB_NEEDS_DOWNLOAD))) {
+		if (needsDownload()) {
 			synchronization.resumeDownload();
 		}
 	}
@@ -164,6 +167,14 @@ public class Job {
 		return getSafeBoolean(getProperty(JOB_IS_DOWNLOADED));
 	}
 	
+	public boolean needsDownload() {
+		return Boolean.parseBoolean(getProperty(JOB_NEEDS_DOWNLOAD));
+	}
+	
+	public boolean needsUpload() {
+		return Boolean.parseBoolean(getProperty(JOB_NEEDS_UPLOAD));
+	}
+	
 	public void uploadFile(String file, ProgressNotifier notifier) {
 		uploadFiles(Arrays.asList(file), notifier);
 	}
@@ -188,7 +199,12 @@ public class Job {
 				String item;
 				progress.startNewFile(totalSizes.get(index));
 				notifier.addItem(item = "Uploading file: " + file.getName());
-				transfer.upload(file);
+				try {
+					transfer.upload(file);
+				} catch (InterruptedIOException e) {
+					notifier.itemDone(item);
+					return;
+				}
 				notifier.itemDone(item);
 				index++;
 			}
@@ -231,7 +247,12 @@ public class Job {
 				String item;
 				progress.addItem(item = fileName);
 				progress.startNewFile(fileSizes.get(idx));
-				transfer.download(fileName, jobDir);
+				try {
+					transfer.download(fileName, jobDir);
+				} catch (InterruptedIOException e) {
+					progress.itemDone(item);
+					return;
+				}
 				progress.itemDone(item);
 				idx++;
 			}

haas-imagej-client/src/main/java/cz/it4i/fiji/haas/data_transfer/PersitentSynchronizationProcess.java → haas-imagej-client/src/main/java/cz/it4i/fiji/haas/data_transfer/PersistentSynchronizationProcess.java

 package cz.it4i.fiji.haas.data_transfer;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -17,12 +21,12 @@ import cz.it4i.fiji.haas_java_client.HaaSFileTransfer;
 import cz.it4i.fiji.haas_java_client.ProgressNotifier;
 import cz.it4i.fiji.haas_java_client.TransferFileProgressForHaaSClient;
 
-public abstract class PersitentSynchronizationProcess<T> {
+public abstract class PersistentSynchronizationProcess<T> {
 
 	private boolean startFinished = true;
 
 	public static final Logger log = LoggerFactory
-			.getLogger(cz.it4i.fiji.haas.data_transfer.PersitentSynchronizationProcess.class);
+			.getLogger(cz.it4i.fiji.haas.data_transfer.PersistentSynchronizationProcess.class);
 
 	private static final TransferFileProgressForHaaSClient DUMMY_FILE_PROGRESS = new TransferFileProgressForHaaSClient(
 			0, HaaSClient.DUMMY_PROGRESS_NOTIFIER);
@@ -30,6 +34,8 @@ public abstract class PersitentSynchronizationProcess<T> {
 	private PersistentIndex<T> index;
 
 	private Queue<T> toProcessQueue = new LinkedBlockingQueue<T>();
+	
+	private Set<Thread> runningTransferThreads = Collections.synchronizedSet(new HashSet<>());
 
 	private SimpleThreadRunner runner;
 
@@ -39,7 +45,7 @@ public abstract class PersitentSynchronizationProcess<T> {
 
 	private ProgressNotifier notifier;
 
-	public PersitentSynchronizationProcess(ExecutorService service,
+	public PersistentSynchronizationProcess(ExecutorService service,
 			Supplier<HaaSFileTransfer> fileTransferSupplier, Runnable processFinishedNotifier, Path indexFile,
 			Function<String, T> convertor) throws IOException {
 		runner = new SimpleThreadRunner(service);
@@ -67,6 +73,7 @@ public abstract class PersitentSynchronizationProcess<T> {
 		toProcessQueue.clear();
 		index.clear();
 		notifier.setCount(-1, -1);
+		runningTransferThreads.forEach(t -> t.interrupt());
 	}
 
 	public void resume() {
@@ -76,25 +83,33 @@ public abstract class PersitentSynchronizationProcess<T> {
 
 	abstract protected Iterable<T> getItems() throws IOException;
 
-	abstract protected void processItem(HaaSFileTransfer tr, T p);
+	abstract protected void processItem(HaaSFileTransfer tr, T p) throws InterruptedIOException;
 
 	private void doProcess(AtomicBoolean reRun) {
+		boolean interrupted = false;
 		try (HaaSFileTransfer tr = fileTransferSupplier.get()) {
 			TransferFileProgressForHaaSClient notifier;
 			tr.setProgress(notifier = getTransferFileProgress(tr));
+			runningTransferThreads.add(Thread.currentThread());
 			while (!toProcessQueue.isEmpty()) {
 				T p = toProcessQueue.poll();
 				String item = p.toString(); 
 				notifier.addItem(item);
-				processItem(tr, p);
-				fileUploaded(p);
+				try {
+					processItem(tr, p);
+					fileUploaded(p);
+				} catch (InterruptedIOException e) {
+					toProcessQueue.clear();
+					interrupted = true;
+				}
 				notifier.itemDone(item);
 				reRun.set(false);
 			}
+			runningTransferThreads.remove(Thread.currentThread());
 			notifier.done();
 		} finally {
 			synchronized (this) {
-				if (startFinished) {
+				if (startFinished && !interrupted && !Thread.interrupted()) {
 					processFinishedNotifier.run();
 				}
 			}

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

 package cz.it4i.fiji.haas.data_transfer;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -37,7 +38,7 @@ public class Synchronization {
 	
 	private PersistentIndex<Path> filesDownloaded;
 
-	private PersitentSynchronizationProcess<Path> uploadProcess;
+	private PersistentSynchronizationProcess<Path> uploadProcess;
 
 	private P_PersistentDownloadProcess downloadProcess;
 	
@@ -92,9 +93,9 @@ public class Synchronization {
 		return !file.getFileName().toString().matches("[.][^.]+") && !filesDownloaded.contains(file);
 	}
 
-	private PersitentSynchronizationProcess<Path> createUploadProcess(Supplier<HaaSFileTransfer> fileTransferSupplier,
+	private PersistentSynchronizationProcess<Path> createUploadProcess(Supplier<HaaSFileTransfer> fileTransferSupplier,
 			ExecutorService service, Runnable uploadFinishedNotifier) throws IOException {
-		return new PersitentSynchronizationProcess<Path>(service, fileTransferSupplier, uploadFinishedNotifier,
+		return new PersistentSynchronizationProcess<Path>(service, fileTransferSupplier, uploadFinishedNotifier,
 				workingDirectory.resolve(FILE_INDEX_TO_UPLOAD_FILENAME), pathResolver) {
 
 			@Override
@@ -106,7 +107,7 @@ public class Synchronization {
 			}
 	
 			@Override
-			protected void processItem(HaaSFileTransfer tr, Path p) {
+			protected void processItem(HaaSFileTransfer tr, Path p) throws InterruptedIOException {
 				UploadingFile uf = new UploadingFileImpl(p);
 				tr.upload(uf);
 			}
@@ -132,7 +133,7 @@ public class Synchronization {
 		return new P_PersistentDownloadProcess(service, fileTransferSupplier, uploadFinishedNotifier);
 	}
 	
-	private class P_PersistentDownloadProcess extends PersitentSynchronizationProcess<String>{
+	private class P_PersistentDownloadProcess extends PersistentSynchronizationProcess<String>{
 
 		private Collection<String> items = Collections.emptyList();
 		
@@ -152,7 +153,7 @@ public class Synchronization {
 		}
 
 		@Override
-		protected void processItem(HaaSFileTransfer tr, String file) {
+		protected void processItem(HaaSFileTransfer tr, String file) throws InterruptedIOException {
 			filesDownloaded.insert(workingDirectory.resolve(file));
 			try {
 				filesDownloaded.storeToFile();

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

@@ -5,12 +5,14 @@ import java.util.LinkedList;
 import java.util.function.BiConsumer;
 import java.util.function.BiPredicate;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Predicate;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javafx.event.EventHandler;
+import javafx.scene.control.CheckMenuItem;
 import javafx.scene.control.ContextMenu;
 import javafx.scene.control.MenuItem;
 import javafx.scene.control.TableColumn;
@@ -21,7 +23,7 @@ public class TableViewContextMenu<T> {
 
 	public final static Logger log = LoggerFactory.getLogger(cz.it4i.fiji.haas.ui.TableViewContextMenu.class);
 
-	private final Collection<P_MenuItem> items = new LinkedList<>();
+	private final Collection<P_Updetable<?>> items = new LinkedList<>();
 	private final Collection<P_MenuItemWithColumnIndex> itemsWithColumnIndex = new LinkedList<>();
 
 	private TableView<T> tableView;
@@ -39,6 +41,11 @@ public class TableViewContextMenu<T> {
 	public void addItem(String text, BiConsumer<T, Integer> eventHandler, BiPredicate<T, Integer> enableHandler) {
 		itemsWithColumnIndex.add(new P_MenuItemWithColumnIndex(text, eventHandler, enableHandler));
 	}
+	
+	public void addItem(String text, Consumer<T> eventHandlerOn, Consumer<T> eventHandlerOff,
+			Predicate<T> enableHandler, Function<T, Boolean> property) {
+		items.add(new P_CheckMenuItem(text, eventHandlerOff, eventHandlerOn, enableHandler, property));
+	}
 
 	private T getRequestedItem() {
 		return tableView.getFocusModel().getFocusedItem();
@@ -82,21 +89,35 @@ public class TableViewContextMenu<T> {
 		return cm;
 	}
 
-	private class P_MenuItem {
-
-		private MenuItem item;
-		private Predicate<T> enableHandler;
-
-		public P_MenuItem(String text, Consumer<T> eventHandler, Predicate<T> enableHandler) {
+	private class P_Updetable<I extends MenuItem> {
+		
+		private I item;
+				private Predicate<T> enableHandler;
+		
+		
+		public P_Updetable(I item, Predicate<T> enableHandler) {
+			this.item = item;
 			this.enableHandler = enableHandler;
-			item = new MenuItem(text);
-			item.setOnAction(e -> eventHandler.accept(getRequestedItem()));
-			getOrCreateContextMenu().getItems().add(item);
+			getOrCreateContextMenu().getItems().add(getItem());
 		}
 
 		public void updateEnable(T selected) {
 			item.setDisable(!enableHandler.test(selected));
 		}
+		
+		protected I getItem() {
+			return item;
+		}
+
+	}
+	
+	private class P_MenuItem extends P_Updetable<MenuItem>{
+		public P_MenuItem(String text, Consumer<T> eventHandler, Predicate<T> enableHandler) {
+			super(new MenuItem(text), enableHandler);
+			
+			getItem().setOnAction(e -> eventHandler.accept(getRequestedItem()));
+			
+		}
 
 	}
 
@@ -119,4 +140,30 @@ public class TableViewContextMenu<T> {
 
 	}
 
+	private class P_CheckMenuItem extends P_Updetable<CheckMenuItem>{
+		
+		private Function<T, Boolean> property;
+
+		public P_CheckMenuItem(String text, Consumer<T> eventHandlerOff, Consumer<T> eventHandlerOn,
+				Predicate<T> enableHandler, Function<T,Boolean> property) {
+			super(new CheckMenuItem(text), enableHandler);
+			this.property = property;
+			getItem().setOnAction(e -> {
+				boolean selected = getItem().isSelected();
+				if(selected) {
+					eventHandlerOn.accept(getRequestedItem());
+				} else {
+					eventHandlerOff.accept(getRequestedItem());
+				}
+			});
+		}
+
+		@Override
+		public void updateEnable(T selected) {
+			super.updateEnable(selected);
+			getItem().setSelected(property.apply(selected));
+		}
+		
+	}
+
 }

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

 package cz.it4i.fiji.haas_java_client;
 
 import java.io.Closeable;
+import java.io.InterruptedIOException;
 import java.nio.file.Path;
 import java.util.List;
 
@@ -13,9 +14,9 @@ public interface HaaSFileTransfer extends Closeable {
 	@Override
 	void close();
 
-	void upload(UploadingFile files);
+	void upload(UploadingFile files) throws InterruptedIOException;
 
-	void download(String files, Path workDirectory);
+	void download(String files, Path workDirectory) throws InterruptedIOException;
 
 	List<Long> obtainSize(List<String> files);
 

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

@@ -3,6 +3,7 @@ package cz.it4i.fiji.haas_java_client;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.nio.file.Path;
 import java.util.LinkedList;
 import java.util.List;
@@ -39,25 +40,30 @@ class HaaSFileTransferImp implements HaaSFileTransfer {
 	}
 
 	@Override
-	public void upload(UploadingFile file) {
+	public void upload(UploadingFile file) throws InterruptedIOException{
 		String destFile = "'" + ft.getSharedBasepath() + "/" + file.getName() + "'";
 		try (InputStream is = file.getInputStream()) {
 			boolean result = scpClient.upload(is, destFile, file.getLength(), file.getLastTime(), progress);
 			if (!result) {
 				throw new HaaSClientException("Uploading of " + file + " to " + destFile + " failed");
 			}
+		} catch(InterruptedIOException e) {
+			throw e;
 		} catch (JSchException | IOException e) {
-			throw new HaaSClientException();
+			throw new HaaSClientException(e);
 		}
 	}
 
 	@Override
-	public void download(String fileName, Path workDirectory) {
+	public void download(String fileName, Path workDirectory) throws InterruptedIOException{
 		try {
 			fileName = fileName.replaceFirst("/", "");
 			Path rFile = workDirectory.resolve(fileName);
 			String fileToDownload = "'" + ft.getSharedBasepath() + "/" + fileName + "'";
 			scpClient.download(fileToDownload, rFile, progress);
+		} catch(InterruptedIOException e) {
+			log.info("upload interrupted flag: " + Thread.currentThread().isInterrupted());
+			throw e;
 		} catch (JSchException | IOException e) {
 			throw new HaaSClientException(e);
 		}
@@ -68,36 +74,6 @@ class HaaSFileTransferImp implements HaaSFileTransfer {
 		this.progress = progress;
 	}
 	
-	/*
-	@Override
-	public void download(Iterable<String> files, Path workDirectory) {
-		List<Long> fileSizes;
-		try {
-			fileSizes = HaaSClient.getSizes(StreamSupport.stream(files.spliterator(), false)
-					.map(filename -> "'" + ft.getSharedBasepath() + "/" + filename + "'").collect(Collectors.toList()),
-					scpClient);
-
-			final long totalFileSize = fileSizes.stream().mapToLong(i -> i.longValue()).sum();
-			TransferFileProgressForHaaSClient progress = new TransferFileProgressForHaaSClient(totalFileSize,
-					HaaSClient.DUMMY_PROGRESS_NOTIFIER);
-			int idx = 0;
-			for (String fileName : files) {
-				fileName = fileName.replaceFirst("/", "");
-				Path rFile = workDirectory.resolve(fileName);
-				String fileToDownload = "'" + ft.getSharedBasepath() + "/" + fileName + "'";
-				String item;
-				progress.addItem(item = fileName);
-				progress.startNewFile(fileSizes.get(idx));
-				scpClient.download(fileToDownload, rFile, progress);
-				progress.itemDone(item);
-				idx++;
-			}
-		} catch (JSchException | IOException e) {
-			throw new HaaSClientException(e);
-		}
-	}
-	*/
-
 	@Override
 	public List<Long> obtainSize(List<String> files) {
 		try {
@@ -128,7 +104,7 @@ class HaaSFileTransferImp implements HaaSFileTransfer {
 		}
 		return result;
 	}
-
+	
 	private String replaceIfFirstFirst(String fileName, String string, String string2) {
 		if (fileName.length() < 0 && fileName.charAt(0) == '/') {
 			fileName = fileName.substring(1);

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

+package cz.it4i.fiji.haas_java_client;
+
+public class LambdaExceptionHandlerWrapper {
+
+	public interface Runnable {
+		void run() throws Exception;
+	}
+
+	public static void wrap(Runnable r) {
+		try {
+			r.run();
+		} catch (Exception e) {
+			//ignore
+		}
+	}
+}

haas-java-client/src/test/java/cz/it4i/fiji/haas_java_client/TestHaaSJavaClient.java

 package cz.it4i.fiji.haas_java_client;
 
+import static cz.it4i.fiji.haas_java_client.LambdaExceptionHandlerWrapper.wrap;
+
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -49,7 +51,7 @@ public class TestHaaSJavaClient {
 			if (info.getState() == JobState.Finished) {
 				try (HaaSFileTransfer fileTransfer = client.startFileTransfer(jobId,
 						HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS)) {
-					client.getChangedFiles(jobId).forEach(file -> fileTransfer.download(file, workDir));
+					client.getChangedFiles(jobId).forEach(file -> wrap(() -> fileTransfer.download(file, workDir)));
 				}
 			}
 			log.info("JobId :" + jobId + ", state" + info.getState());

haas-java-client/src/test/java/cz/it4i/fiji/haas_java_client/TestHaaSJavaClientWithSPIM.java

@@ -14,24 +14,24 @@ import org.slf4j.LoggerFactory;
 
 import cz.it4i.fiji.haas_java_client.HaaSClient.SynchronizableFiles;
 import cz.it4i.fiji.haas_java_client.proxy.JobFileContentExt;
+import static cz.it4i.fiji.haas_java_client.LambdaExceptionHandlerWrapper.wrap;
 
 public class TestHaaSJavaClientWithSPIM {
 
 	private static Logger log = LoggerFactory.getLogger(cz.it4i.fiji.haas_java_client.TestHaaSJavaClientWithSPIM.class);
-	
+
 	public static void main(String[] args) throws ServiceException, IOException {
 		HaaSClient client = new HaaSClient(TestingConstants.getSettings(2, 9600, 6l, "DD-17-31"));
 		Path baseDir = Paths.get("/home/koz01/Work/vyzkumnik/fiji/work/aaa");
 
-		long jobId = client.createJob( "TestOutRedirect",
-				Collections.emptyList());
-		
-		try(HaaSFileTransfer tr = client.startFileTransfer(jobId, HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS)) {
+		long jobId = client.createJob("TestOutRedirect", Collections.emptyList());
+
+		try (HaaSFileTransfer tr = client.startFileTransfer(jobId, HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS)) {
 			StreamSupport.stream(getAllFiles(baseDir.resolve("spim-data")).spliterator(), false)
-					.map(UploadingFileImpl::new).forEach(f -> tr.upload(f));
+					.map(UploadingFileImpl::new).forEach(f -> wrap(() -> tr.upload(f)));
 		}
 		client.submitJob(jobId);
-		
+
 		Path workDir = baseDir.resolve("" + jobId);
 		if (!Files.isDirectory(workDir)) {
 			Files.createDirectories(workDir);
@@ -53,8 +53,9 @@ public class TestHaaSJavaClientWithSPIM {
 			}
 			client.downloadPartsOfJobFiles(jobId, taskFileOffset).forEach(jfc -> showJFC(jfc));
 			if (info.getState() == JobState.Finished) {
-				try (HaaSFileTransfer fileTransfer = client.startFileTransfer(jobId, HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS)) {
-					client.getChangedFiles(jobId).forEach(file -> fileTransfer.download(file, workDir));
+				try (HaaSFileTransfer fileTransfer = client.startFileTransfer(jobId,
+						HaaSClient.DUMMY_TRANSFER_FILE_PROGRESS)) {
+					client.getChangedFiles(jobId).forEach(file -> wrap(() -> fileTransfer.download(file, workDir)));
 				}
 
 			}

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

@@ -434,6 +434,18 @@ public class BenchmarkJobManager {
 		public boolean isUploaded() {
 			return job.isUploaded();
 		}
+
+		public void stopDownload() {
+			job.stopDownloadData();
+		}
+
+		public boolean needsDownload() {
+			return job.needsDownload();
+		}
+
+		public boolean needsUpload() {
+			return job.needsUpload();
+		}
 	}
 
 	public BenchmarkJobManager(BenchmarkSPIMParameters params) throws IOException {

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

@@ -18,9 +18,9 @@ public class UpdatableBenchmarkJob extends UpdatableObservableValue<BenchmarkJob
 			.getLogger(cz.it4i.fiji.haas_spim_benchmark.core.UpdatableBenchmarkJob.class);
 
 	private P_TransferProgress downloadProgress = new P_TransferProgress(val -> getValue().setDownloaded(val),
-			() -> getValue().isDownloaded());
+			() -> getValue().isDownloaded(), () -> getValue().needsDownload());
 	private P_TransferProgress uploadProgress = new P_TransferProgress(val -> getValue().setUploaded(val),
-			() -> getValue().isUploaded());
+			() -> getValue().isUploaded(), () -> getValue().needsUpload());
 	private Executor executor;
 
 	public interface TransferProgress {
@@ -58,28 +58,29 @@ public class UpdatableBenchmarkJob extends UpdatableObservableValue<BenchmarkJob
 
 	@Override
 	protected void fireValueChangedEvent() {
-		executor.execute(() -> super.fireValueChangedEvent());
+		executor.execute(() ->  {
+			super.fireValueChangedEvent();
+		});
 	}
 
 	private class P_TransferProgress implements Progress, TransferProgress {
 
-		private boolean working;
-		// private boolean done;
 		private long start;
 		private Long remainingMiliseconds;
 		private Float remainingPercents;
 		private Supplier<Boolean> doneStatusSupplier;
 		private Consumer<Boolean> doneStatusConsumer;
+		private Supplier<Boolean> workingSupplier;
 
-		public P_TransferProgress(Consumer<Boolean> doneStatusConsumer, Supplier<Boolean> doneStatusSupplier) {
+		public P_TransferProgress(Consumer<Boolean> doneStatusConsumer, Supplier<Boolean> doneStatusSupplier, Supplier<Boolean> workingSupplier) {
 			this.doneStatusConsumer = doneStatusConsumer;
 			this.doneStatusSupplier = doneStatusSupplier;
+			this.workingSupplier = workingSupplier;
 		}
 
 		@Override
 		public synchronized void setCount(int count, int total) {
 			if (total < -1) {
-				working = false;
 				remainingMiliseconds = null;
 				remainingPercents = null;
 			} else {
@@ -92,20 +93,14 @@ public class UpdatableBenchmarkJob extends UpdatableObservableValue<BenchmarkJob
 
 		@Override
 		public synchronized void addItem(Object item) {
-			if (!working) {
-				setDone(false);
-				working = true;
-				start = System.currentTimeMillis();
-			}
+			setDone(false);
+			start = System.currentTimeMillis();
+		
 			fireValueChangedEvent();
 		}
 
 		@Override
 		public synchronized void done() {
-			if (working) {
-				setDone(true);
-			}
-			working = false;
 			remainingMiliseconds = 0l;
 			remainingPercents = 0.f;
 			fireValueChangedEvent();
@@ -113,7 +108,7 @@ public class UpdatableBenchmarkJob extends UpdatableObservableValue<BenchmarkJob
 
 		@Override
 		public synchronized boolean isWorking() {
-			return working;
+			return workingSupplier.get();
 		}
 
 		@Override

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

@@ -110,13 +110,19 @@ public class BenchmarkSPIMController extends BorderPane implements CloseableCont
 						|| j.getState() == JobState.Failed || j.getState() == JobState.Canceled));
 
 		menu.addItem("Upload data", job -> executeWSCallAsync("Uploading data", p -> job.getValue().startUpload()),
-				job -> JavaFXRoutines.notNullValue(job, j -> !EnumSet.of(JobState.Running).contains(j.getState())));
+				job -> executeWSCallAsync("Stop uploading data", p -> job.getValue().stopUpload()),
+				job -> JavaFXRoutines.notNullValue(job, j -> !EnumSet.of(JobState.Running).contains(j.getState())),
+				job -> registry.get(job.getValue()).getUploadProgress().isWorking());
+		
 		
 		menu.addItem("Download result",
 				job -> executeWSCallAsync("Downloading data", p -> job.getValue().startDownload()),
+				job -> executeWSCallAsync("Stop downloading data", p -> job.getValue().stopDownload()),
 				job -> JavaFXRoutines.notNullValue(job,
 						j -> EnumSet.of(JobState.Failed, JobState.Finished, JobState.Canceled).contains(j.getState())
-								&& !j.canBeDownloaded()));
+								&& j.canBeDownloaded()),
+				job -> registry.get(job.getValue()).getDownloadProgress().isWorking());
+
 		menu.addItem("Download statistics",
 				job -> executeWSCallAsync("Downloading data", p -> job.getValue().downloadStatistics(p)),
 				job -> JavaFXRoutines.notNullValue(job, j -> j.getState() == JobState.Finished));

java-scpclient/pom.xml

@@ -32,5 +32,18 @@
 			<artifactId>jsch</artifactId>
 			<version>0.1.54</version>
 		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+			<version>1.7.25</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-jdk14 -->
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-jdk14</artifactId>
+			<version>1.7.25</version>
+			<optional>true</optional>
+		</dependency>
 	</dependencies>
 </project>

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

@@ -3,14 +3,19 @@ package cz.it4i.fiji.scpclient;
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.nio.channels.ClosedByInterruptException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.stream.Collectors;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.jcraft.jsch.Channel;
 import com.jcraft.jsch.ChannelExec;
 import com.jcraft.jsch.ChannelSftp;
@@ -24,6 +29,9 @@ import com.jcraft.jsch.UserInfo;
 
 public class ScpClient implements Closeable {
 	
+	public static final Logger log = LoggerFactory.getLogger(cz.it4i.fiji.scpclient.ScpClient.class);
+	
+	
 	private String hostName;
 	private String username;
 	private JSch jsch = new JSch();
@@ -237,6 +245,9 @@ public class ScpClient implements Closeable {
 			}
 			out.close();
 
+		} catch (ClosedByInterruptException e) {
+			Thread.interrupted();
+			throw new InterruptedIOException();
 		} finally {
 			channel.disconnect();
 		}