+ public void test_cancel_noBlockingWhenBaseRunnableNotScheduled() throws Exception{
+ //nothing is running so this should return immediately without TimeoutException
+ asyncTaskHelper.cancelBaseActionRunnable(AppcOam.RPC.stop , AppcOamStates.Started , 1, TimeUnit.MILLISECONDS);
+ }
+
+
+
+ /**
+ * Test {@link AsyncTaskHelper#cancelBaseActionRunnable(AppcOam.RPC, AppcOamStates, long, TimeUnit)}}
+ * Test cancel does blocks until BaseRunnable is done scheduled
+ */
+ @Test()
+ public void test_cancel_BlockingWhenBaseRunnableNotDone() throws Exception {
+
+
+ final ExecuteTest et = new ExecuteTest();
+ et.isContinuous = true;
+ et.isIgnoreInterrupt = true;
+ asyncTaskHelper.scheduleBaseRunnable(
+ et::test
+ , s -> {
+ }
+ , initialDelayMillis
+ , delayMillis
+ );
+
+ Assert.assertTrue("It should be running", waitFor(et::isExecuting, 1000));
+
+
+ //we should get a timeout
+ try {
+ asyncTaskHelper.cancelBaseActionRunnable(
+ AppcOam.RPC.stop,
+ AppcOamStates.Started,
+ 1,
+ TimeUnit.MILLISECONDS);
+ Assert.fail("Should have gotten TimeoutException");
+ } catch (TimeoutException e) {
+ //just ignore as it is expected
+ }
+
+
+ //release the test thread
+ et.isContinuous = false;
+
+
+ //we should not get a timeout
+ asyncTaskHelper.cancelBaseActionRunnable(
+ AppcOam.RPC.stop,
+ AppcOamStates.Started,
+ 1000,
+ TimeUnit.MILLISECONDS);
+
+ }
+
+
+
+ /**
+ * Test {@link AsyncTaskHelper#cancelBaseActionRunnable(AppcOam.RPC, AppcOamStates, long, TimeUnit)}}
+ * Test cancel does not block when BaseRunnable is not scheduled
+ */
+ @Test
+ public void test_BaseRunnableCancelCallback() throws Exception{
+
+ AtomicReference<AppcOam.RPC> cancelCallback = new AtomicReference<>(null);
+
+ final ExecuteTest et = new ExecuteTest();
+ et.isContinuous = true;
+ Future<?> future = asyncTaskHelper.scheduleBaseRunnable(
+ et::test
+ , cancelCallback::set
+ , initialDelayMillis
+ , delayMillis
+ );
+
+ Assert.assertTrue("It should be running", waitFor(et::isExecuting, 1000));
+ Assert.assertTrue("It should be running", waitForNot(future::isDone, 1000));
+
+
+ try {
+ asyncTaskHelper.cancelBaseActionRunnable(
+ AppcOam.RPC.stop,
+ AppcOamStates.Started,
+ 1,
+ TimeUnit.MILLISECONDS);
+ Assert.fail("Should have gotten TimeoutException");
+ } catch (TimeoutException e) {
+ //just ignore as it is expected
+ }
+
+
+ Assert.assertEquals("Unexpected rpc in call back",AppcOam.RPC.stop,cancelCallback.get());
+ }
+
+
+
+
+
+
+
+
+ /**
+ * @return true if the negation of the expected value is returned from the supplier within the specified
+ * amount of time
+ */
+ private static boolean waitForNot(Supplier<Boolean> s,long timeoutMillis)throws Exception{
+ return waitFor(()->!s.get(),timeoutMillis);
+ }
+
+
+ /**
+ * @return true if the expected value is returned from the supplier within the specified
+ * amount of time
+ */
+ private static boolean waitFor(Supplier<Boolean> s,long timeoutMillis) throws Exception {
+ long timeout = TimeUnit.MILLISECONDS.toMillis(timeoutMillis);
+ long expiryTime = System.currentTimeMillis() + timeout;
+ long elapsedTime;
+ while(!s.get()){
+ elapsedTime = expiryTime - System.currentTimeMillis();
+ if(elapsedTime < 1) {
+ break;
+ }
+ Thread.sleep(10);
+ }
+ return s.get();
+ }
+
+
+ /**
+ * This class is used control a thread executed in th {@link #test()}
+ */
+ @SuppressWarnings("unused")
+ private static class ExecuteTest {
+
+
+ /** A fail safe to insure this TEst does not run indefinitely */
+ private final long EXPIRY_TIME = System.currentTimeMillis() + 10000;
+
+
+
+ /** A thread sets this value to true when it has completed the execution the of executes {@link #test()} */
+ private volatile boolean isExecuted = false;
+
+ /**
+ * A thread sets this value to true when it is actively executing {@link #test()} and back to false when
+ * it is not
+ */
+ private volatile boolean isExecuting = false;
+
+ /**
+ * While this value is true, a thread will not be allowed to return from {@link #test()} It will simulate a
+ * long execution.
+ */
+ private volatile boolean isContinuous = false;
+
+ /**
+ * When this value is set to true, an ongoing simulation of a long execution of {@link #test()} cannot be force
+ * to abort via a {@link Thread#interrupt()}
+ */
+ private volatile boolean isIgnoreInterrupt = false;
+
+
+
+ /** Use to send a signal to the thread executing {@link #notifyTestExcuted(long)} */
+ private Semaphore inner = new Semaphore(0);
+
+ /** Use to send a signal to the thread executing {@link #waitForTestExec(long)} */
+ private Semaphore outer = new Semaphore(0);
+
+ /** The {@link Future} of the Thread executing {@link #test()}*/
+ private volatile Future<?> future;
+
+ /**
+ * When set the Thread executing {@link #test()} will cancel itself
+ * @param future - The {@link Future} of the Thread executing {@link #test()}
+ */
+ private void cancelSelfOnNextExecution(Future<?> future) {
+ this.future = future;
+ }
+
+
+ private boolean isExecuted() {
+ return isExecuted;
+ }
+
+ private boolean isExecuting() {
+ return isExecuting;
+ }
+
+
+ private boolean isContinuous() {
+ return isContinuous;
+ }
+
+
+ private boolean isIgnoreInterrupt() {
+ return isIgnoreInterrupt;
+ }
+
+
+
+ /**
+ * The thread executing this method if blocked from returning until the thread executing
+ * {@link #test()} invokes {@link #notifyTestExcuted(long)} or the specified time elapses
+ * @param timeoutMillis - the amount of time to wait for a execution iteration.
+ * @return true if the Thread is released because of an invocation of {@link #notifyTestExcuted(long)}
+ * @throws InterruptedException - If the Caller thread is interrupted.
+ */
+ private boolean waitForTestExec(long timeoutMillis) throws InterruptedException {
+ inner.release();
+ return outer.tryAcquire(timeoutMillis,TimeUnit.MILLISECONDS);
+ }
+
+
+ /**
+ * Test simulator
+ * @return Always returns true.
+ */
+ private Boolean test() {
+ isTestExpired();
+ System.out.println("started");
+ isExecuting = true;
+ try {
+ if (future != null) {
+ future.cancel(false);
+ }
+ if(!isContinuous){
+ notifyTestExcuted(1);
+ }
+
+ while(isContinuous){
+ notifyTestExcuted(100);
+ isTestExpired();
+ }
+
+ } finally {
+ isExecuting = false;
+ isExecuted = true;
+ }
+ return true;
+ }
+
+
+ /** @throws RuntimeException if the test has bee running too long */
+ private void isTestExpired(){
+ if(System.currentTimeMillis() > EXPIRY_TIME){
+ throw new RuntimeException("Something went wrong the test expired.");
+ }
+ }
+
+
+ /**
+ * The thread executing {@link #test()} if blocked from returning until another thread invokes
+ * {@link #waitForTestExec(long)} or the specified time elapses
+ * @param timeoutMillis - the amount of time to wait for a execution iteration.
+ * @return true if the Thread is released because of an invocation of {@link #waitForTestExec(long)}
+ */
+ private boolean notifyTestExcuted(long timeoutMillis){
+ try {
+ boolean acquire = inner.tryAcquire(timeoutMillis,TimeUnit.MILLISECONDS);
+ if(acquire){
+ outer.release();
+ System.out.println("release");
+ }
+ } catch (InterruptedException e) {
+ if(!isIgnoreInterrupt){
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+
+
+ static class Sub {
+ ExecuteTest et = new ExecuteTest();
+ Future<?> future = null;