Refactor ExternalProcessHandler. Refactor and fix DataUpdateFuture.

Fix CompositeCancelMonitor

plugin_tooling/src-lang/melnorme/lang/utils/concurrency/ConcurrentlyDerivedData.java

@@ -14,13 +14,12 @@
 import static melnorme.utilbox.core.Assert.AssertNamespace.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import melnorme.lang.tooling.common.ops.IOperationMonitor;
+import melnorme.utilbox.concurrency.AsyncSupplier;
 import melnorme.utilbox.concurrency.CancellableTask;
-import melnorme.utilbox.concurrency.NonCancellableFuture;
 import melnorme.utilbox.concurrency.OperationCancellation;
 import melnorme.utilbox.core.fntypes.CallableX;
 import melnorme.utilbox.fields.ListenerListHelper;
@@ -171,48 +170,43 @@ protected final void doRun() {
 
 	/* -----------------  ----------------- */
 
-
-	protected final DataUpdateFuture asFuture = new DataUpdateFuture();
-
 	/** 
-	 * @return a {@link Future} that can be used to wait for the first non-stale data that becomes available.
+	 * @return a {@link AsyncSupplier} that can be used to wait for the first non-stale data that becomes available.
 	 */
-	public DataUpdateFuture asFuture() {
-		return asFuture;
+	public DataUpdateFuture getSupplierForNextUpdate() {
+		return new DataUpdateFuture();
 	}
 
 	public DATA awaitUpdatedData() throws InterruptedException {
-		return asFuture().awaitResult();
+		return getSupplierForNextUpdate().awaitResult();
 	}
 
 	public DATA awaitUpdatedData(IOperationMonitor om) throws OperationCancellation {
-		return asFuture().awaitData(om);
+		return getSupplierForNextUpdate().awaitResult(om);
 	}
 
-	public class DataUpdateFuture implements NonCancellableFuture<DATA> {
-		@Override
-		public boolean isTerminated() {
-			return !isStale();
-		}
+	public class DataUpdateFuture implements AsyncSupplier<DATA> {
 
-		@Override
-		public DATA awaitResult() throws InterruptedException {
+		public void awaitTermination() throws InterruptedException {
 			getLatchForUpdateTask().await();
-			return data;
 		}
 
-		@Override
-		public DATA awaitResult(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+		public void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
 			boolean success = getLatchForUpdateTask().await(timeout, unit);
 			if(!success) {
 				throw new TimeoutException();
 			}
+		}
+
+		@Override
+		public DATA awaitResult() throws InterruptedException {
+			awaitTermination();
 			return data;
 		}
 
 		@Override
-		public DATA getResult_forSuccessfulyCompleted() {
-			assertTrue(isCompletedSuccessfully());
+		public DATA awaitResult(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+			awaitTermination(timeout, unit);
 			return data;
 		}
 

plugin_tooling/src-util/melnorme/utilbox/concurrency/AbstractFuture2.java

@@ -50,14 +50,13 @@ protected void handleCancellation() {
 	}
 
 	@Override
-	public RET awaitResult() throws InterruptedException, OperationCancellation {
-		return completableResult.awaitResult();
+	public void awaitTermination() throws InterruptedException {
+		completableResult.awaitTermination();
 	}
 
 	@Override
-	public RET awaitResult(long timeout, TimeUnit unit) 
-			throws InterruptedException, TimeoutException, OperationCancellation {
-		return completableResult.awaitResult(timeout, unit);
+	public void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+		completableResult.awaitTermination(timeout, unit);
 	}
 
 	@Override

plugin_tooling/src-util/melnorme/utilbox/concurrency/AbstractRunnableFuture2.java

@@ -11,7 +11,7 @@
 package melnorme.utilbox.concurrency;
 
 public abstract class AbstractRunnableFuture2<RET> extends AbstractFuture2<RET> 
-	implements IRunnableFuture2<RET>	
+	implements IRunnableFuture2<RET>
 {
 
 	public AbstractRunnableFuture2() {
@@ -50,6 +50,12 @@ protected void internalTaskRun() {
 
 	protected abstract RET internalInvoke();
 
+
+	public void completeWithResult(RET result) {
+		cancellableTask.markExecuted();
+		completableResult.setResult(result);
+	}
+
 	/* -----------------  ----------------- */
 
 	@Override

plugin_tooling/src-util/melnorme/utilbox/concurrency/AsyncSupplier.java

@@ -0,0 +1,58 @@
+/*******************************************************************************
+ * Copyright (c) 2016 Bruno Medeiros and other Contributors.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *     Bruno Medeiros - initial API and implementation
+ *******************************************************************************/
+package melnorme.utilbox.concurrency;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Interface for a blocking/asynchrounous supplier.
+ * There is no guarantee, with this interface alone, that the supplier is the same for multiple calls
+ * of awaitResult
+ *
+ * @param <RESULT>
+ */
+public interface AsyncSupplier<RESULT> {
+
+	RESULT awaitResult()
+			throws OperationCancellation, InterruptedException;
+
+	RESULT awaitResult(long timeout, TimeUnit unit)
+			throws OperationCancellation, InterruptedException, TimeoutException;
+
+	/** Same as {@link #awaitResult()}, 
+	 * but throw InterruptedException as an OperationCancellation. */
+	default RESULT awaitResult2() throws OperationCancellation {
+		try {
+			return awaitResult();
+		} catch(InterruptedException e) {
+			throw new OperationCancellation();
+		}
+	}
+
+	/* -----------------  ----------------- */
+
+	default RESULT awaitResult(ICancelMonitor cm) throws OperationCancellation {
+
+		while(true) {
+			cm.checkCancellation();
+
+			try {
+				return this.awaitResult(100, TimeUnit.MILLISECONDS);
+			} catch(InterruptedException e) {
+				throw new OperationCancellation();
+			} catch(TimeoutException e) {
+				continue;
+			}
+		}
+	}
+
+}

plugin_tooling/src-util/melnorme/utilbox/concurrency/BasicFuture.java

@@ -32,7 +32,7 @@
  * @see Future2
  * 
  */
-public interface BasicFuture<RESULT> {
+public interface BasicFuture<RESULT> extends AsyncSupplier<RESULT> {
 
     /** @return true if execution of this future has terminated already. */
     boolean isTerminated();
@@ -47,22 +47,10 @@ default boolean isCompletedSuccessfully() {
 
 
     /* -----------------  ----------------- */
-
-	RESULT awaitResult() 
-			throws OperationCancellation, InterruptedException;
-
-	RESULT awaitResult(long timeout, TimeUnit unit) 
-			throws OperationCancellation, InterruptedException, TimeoutException;
 
-	/** Same as {@link #awaitResult()}, 
-	 * but throw InterruptedException as an OperationCancellation. */
-	default RESULT awaitResult2() throws OperationCancellation {
-		try {
-			return awaitResult();
-		} catch(InterruptedException e) {
-			throw new OperationCancellation();
-		}
-	}
+	void awaitTermination() throws InterruptedException;
+
+	void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException;
 
 	/**
 	 * It is only legal to call this method if the future has terminated already.
@@ -84,4 +72,20 @@ default RESULT getResult_forTerminated() throws OperationCancellation {
 	 */
 	RESULT getResult_forSuccessfulyCompleted();
 
+	/* -----------------  ----------------- */
+
+	@Override
+	default RESULT awaitResult() 
+			throws OperationCancellation, InterruptedException {
+		awaitTermination();
+		return getResult_forTerminated();
+	}
+
+	@Override
+	default RESULT awaitResult(long timeout, TimeUnit unit) 
+			throws OperationCancellation, InterruptedException, TimeoutException {
+		awaitTermination(timeout, unit);
+		return getResult_forTerminated();
+	}
+
 }

plugin_tooling/src-util/melnorme/utilbox/concurrency/CancellableTask.java

@@ -70,7 +70,7 @@ public final void run() {
 			}
 
 			// Can only run once
-			checkExecuted();
+			markExecuted();
 			assertTrue(runningThread == null);
 			runningThread = Thread.currentThread();
 		}
@@ -96,7 +96,7 @@ public final void run() {
 		}
 	}
 
-	protected void checkExecuted() {
+	public void markExecuted() {
 		assertTrue(executed == false); // Check that run is only called once
 		executed = true;
 	}

plugin_tooling/src-util/melnorme/utilbox/concurrency/CompletableResult.java

@@ -37,11 +37,9 @@
  * as opposed to getting silently swallowed/ignored by an executor worker thread. 
  *
  */
-public class CompletableResult<DATA> 
-	implements BasicFuture<DATA>
-{
+public class CompletableResult<DATA> implements BasicFuture<DATA> {
 
-	protected final CountDownLatch completionLatch = new CountDownLatch(1);
+	protected final CountDownLatch terminationLatch = new CountDownLatch(1);
 	protected final Object lock = new Object();
 
     protected volatile ResultStatus status = ResultStatus.NOT_TERMINATED;
@@ -59,17 +57,32 @@ public boolean isTerminated() {
 		return status != ResultStatus.NOT_TERMINATED;
 	}
 
-	public boolean isCompleted() {
-		return isTerminated();
-	}
-
 	@Override
 	public boolean isCancelled() {
 		return status == ResultStatus.CANCELLED;
 	}
 
-	public CountDownLatch getCompletionLatch() {
-		return completionLatch;
+	protected CountDownLatch getTerminationLatch() {
+		return terminationLatch;
+	}
+
+	@Override
+	public void awaitTermination() throws InterruptedException {
+		if(isTerminated()) {
+			return; // Early check so that InterruptedException is not thrown
+		}
+		getTerminationLatch().await();
+	}
+
+	@Override
+	public void awaitTermination(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+		if(isTerminated()) {
+			return; // Early check so that InterruptedException is not thrown
+		}
+		boolean success = getTerminationLatch().await(timeout, unit);
+		if(!success) {
+			throw new TimeoutException();
+		}
 	}
 
 	public void setResultFromCallable(CallableX<DATA, RuntimeException> resultCallable) {
@@ -104,7 +117,7 @@ protected void doSetResult(DATA result, RuntimeException re) {
 			this.result = result;
 			this.resultRuntimeException = re;
 			status = ResultStatus.RESULT_SET;
-			completionLatch.countDown();
+			terminationLatch.countDown();
 		}
 	}
 
@@ -121,7 +134,7 @@ public boolean setCancelledResult() {
 				return false;
 			} else {
 				status = ResultStatus.CANCELLED;
-				completionLatch.countDown();
+				terminationLatch.countDown();
 				return true;
 			}
 		}
@@ -131,37 +144,6 @@ protected void handleReSetResult() {
 		throw assertFail();
 	}
 
-	public void awaitCompletion() throws InterruptedException {
-		if(isCompleted()) {
-			return; // Early check so that InterruptedException is not thrown
-		}
-		completionLatch.await();
-	}
-
-	public void awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
-		if(isCompleted()) {
-			return; // Early check so that InterruptedException is not thrown
-		}
-		boolean success = completionLatch.await(timeout, unit);
-		if(!success) {
-			throw new TimeoutException();
-		}
-	}
-
-	@Override
-	public DATA awaitResult() 
-			throws OperationCancellation, InterruptedException {
-		awaitCompletion();
-		return getResult_forTerminated();
-	}
-
-	@Override
-	public DATA awaitResult(long timeout, TimeUnit unit) 
-			throws OperationCancellation, InterruptedException, TimeoutException {
-		awaitCompletion(timeout, unit);
-		return getResult_forTerminated();
-	}
-
 	@Override
 	public DATA getResult_forSuccessfulyCompleted() {
 		assertTrue(isCompletedSuccessfully());

plugin_tooling/src-util/melnorme/utilbox/concurrency/ICancelMonitor.java

@@ -66,7 +66,7 @@ public CompositeCancelMonitor(ICancelMonitor parentCancelMonitor) {
 
 		@Override
 		public boolean isCancelled() {
-			return super.isCancelled() && parentCancelMonitor.isCancelled();
+			return super.isCancelled() || parentCancelMonitor.isCancelled();
 		}
 	}