brooklyn-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From henev...@apache.org
Subject [03/13] git commit: Converts task tests from groovy to java
Date Fri, 18 Jul 2014 16:34:12 GMT
Converts task tests from groovy to java


Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/35363d61
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/35363d61
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/35363d61

Branch: refs/heads/master
Commit: 35363d613d4b09caa1bd067f0a2b47e10354611c
Parents: 6325e50
Author: Aled Sage <aled.sage@gmail.com>
Authored: Wed Jul 16 23:47:35 2014 +0100
Committer: Aled Sage <aled.sage@gmail.com>
Committed: Thu Jul 17 11:28:44 2014 +0100

----------------------------------------------------------------------
 .../util/task/BasicTaskExecutionTest.groovy     | 370 ---------------
 .../util/task/BasicTaskExecutionTest.java       | 460 +++++++++++++++++++
 .../util/task/ScheduledExecutionTest.groovy     | 148 ------
 .../util/task/ScheduledExecutionTest.java       | 175 +++++++
 .../task/SingleThreadedSchedulerTest.groovy     | 154 -------
 .../util/task/SingleThreadedSchedulerTest.java  | 192 ++++++++
 6 files changed, 827 insertions(+), 672 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.groovy b/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.groovy
deleted file mode 100644
index 1d104a8..0000000
--- a/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.groovy
+++ /dev/null
@@ -1,370 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package brooklyn.util.task;
-
-import static org.testng.Assert.*
-
-import java.util.concurrent.Callable
-import java.util.concurrent.CancellationException
-import java.util.concurrent.CountDownLatch
-import java.util.concurrent.TimeUnit
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod
-import org.testng.annotations.Test
-
-import brooklyn.management.ExecutionManager
-import brooklyn.management.Task
-import brooklyn.test.TestUtils
-
-import com.google.common.base.Throwables
-
-/**
- * Test the operation of the {@link BasicTask} class.
- *
- * TODO clarify test purpose
- */
-public class BasicTaskExecutionTest {
-    private static final Logger log = LoggerFactory.getLogger(BasicTaskExecutionTest.class)
- 
-    private static final int TIMEOUT_MS = 10*1000
-    
-    private BasicExecutionManager em;
-    private Map data;
-
-    @BeforeMethod
-    public void setUp() {
-        em = new BasicExecutionManager("mycontext");
-//        assertTrue em.allTasks.isEmpty()
-        data = Collections.synchronizedMap(new HashMap())
-        data.clear()
-    }
-    
-    @AfterMethod(alwaysRun=true)
-    public void tearDown() throws Exception {
-        if (em != null) em.shutdownNow();
-    }
-    
-    @Test
-    public void runSimpleBasicTask() {
-        data.clear()
-        BasicTask t = [ { data.put(1, "b") } ]
-        data.put(1, "a")
-        BasicTask t2 = em.submit tag:"A", t
-        assertEquals("a", t.get())
-        assertEquals("b", data.get(1))
-    }
-    
-    @Test
-    public void runSimpleRunnable() {
-        data.clear()
-        data.put(1, "a")
-        BasicTask t = em.submit tag:"A", new Runnable() { public void run() { data.put(1, "b") } }
-        assertEquals(null, t.get())
-        assertEquals("b", data.get(1))
-    }
-
-    @Test
-    public void runSimpleCallable() {
-        data.clear()
-        data.put(1, "a")
-        BasicTask t = em.submit tag:"A", new Callable() { public Object call() { data.put(1, "b") } }
-        assertEquals("a", t.get())
-        assertEquals("b", data.get(1))
-    }
-
-    @Test
-    public void runBasicTaskWithWaits() {
-        CountDownLatch signalStarted = new CountDownLatch(1);
-        CountDownLatch allowCompletion = new CountDownLatch(1);
-        data.clear()
-        BasicTask t = [ {
-            def result = data.put(1, "b")
-            signalStarted.countDown();
-            assertTrue(allowCompletion.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
-            result
-        } ]
-        data.put(1, "a")
-
-        BasicTask t2 = em.submit tag:"A", t
-        assertEquals(t, t2)
-        assertFalse(t.isDone())
-        
-        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
-        assertEquals("b", data.get(1))
-        assertFalse(t.isDone())
-        
-        log.debug "runBasicTaskWithWaits, BasicTask status: {}", t.getStatusDetail(false)
-        
-        TestUtils.executeUntilSucceeds { t.getStatusDetail(false).toLowerCase().contains("waiting") }
-        // "details="+t.getStatusDetail(false))
-        
-        allowCompletion.countDown();
-        assertEquals("a", t.get())
-    }
-
-    @Test
-    public void runMultipleBasicTasks() {
-        data.clear()
-        data.put(1, 1)
-        BasicExecutionManager em = []
-        2.times { em.submit tag:"A", new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } }) }
-        2.times { em.submit tag:"B", new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } }) }
-        int total = 0;
-        em.getTaskTags().each {
-                log.debug "tag {}", it
-                em.getTasksWithTag(it).each {
-                    log.debug "BasicTask {}, has {}", it, it.get()
-                    total += it.get()
-                }
-            }
-        assertEquals(10, total)
-        //now that all have completed:
-        assertEquals(5, data.get(1))
-    }
-
-    @Test
-    public void runMultipleBasicTasksMultipleTags() {
-        data.clear()
-        data.put(1, 1)
-        Collection<Task> tasks = []
-        tasks += em.submit tag:"A", new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } })
-        tasks += em.submit tags:["A","B"], new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } })
-        tasks += em.submit tags:["B","C"], new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } })
-        tasks += em.submit tags:["D"], new BasicTask({ synchronized(data) { data.put(1, data.get(1)+1) } })
-        int total = 0;
-
-        tasks.each { Task t ->
-                log.debug "BasicTask {}, has {}", t, t.get()
-                total += t.get()
-            }
-        assertEquals(10, total)
- 
-        //now that all have completed:
-        assertEquals data.get(1), 5
-        assertEquals em.getTasksWithTag("A").size(), 2
-        assertEquals em.getTasksWithAnyTag(["A"]).size(), 2
-        assertEquals em.getTasksWithAllTags(["A"]).size(), 2
-
-        assertEquals em.getTasksWithAnyTag(["A", "B"]).size(), 3
-        assertEquals em.getTasksWithAllTags(["A", "B"]).size(), 1
-        assertEquals em.getTasksWithAllTags(["B", "C"]).size(), 1
-        assertEquals em.getTasksWithAnyTag(["A", "D"]).size(), 3
-    }
-
-    @Test
-    public void testGetTaskById() {
-        Task t = new BasicTask({ /*no-op*/ })
-        em.submit tag:"A",t
-        assertEquals(em.getTask(t.id), t);
-    }
-
-    @Test
-    public void testRetrievingTasksWithTagsReturnsExpectedTask() {
-        Task t = new BasicTask({ /*no-op*/ })
-        em.submit tag:"A",t
-        t.get();
-
-        assertEquals(em.getTasksWithTag("A"), [t]);
-        assertEquals(em.getTasksWithAnyTag(["A"]), [t]);
-        assertEquals(em.getTasksWithAnyTag(["A","B"]), [t]);
-        assertEquals(em.getTasksWithAllTags(["A"]), [t]);
-    }
-
-    @Test
-    public void testRetrievingTasksWithTagsExcludesNonMatchingTasks() {
-        Task t = new BasicTask({ /*no-op*/ })
-        em.submit tag:"A",t
-        t.get();
-
-        assertEquals(em.getTasksWithTag("B"), []);
-        assertEquals(em.getTasksWithAnyTag(["B"]), []);
-        assertEquals(em.getTasksWithAllTags(["A","B"]), []);
-    }
-    
-    @Test
-    public void testRetrievingTasksWithMultipleTags() {
-        Task t = new BasicTask({ /*no-op*/ })
-        em.submit tags:["A","B"], t
-        t.get();
-
-        assertEquals(em.getTasksWithTag("A"), [t]);
-        assertEquals(em.getTasksWithTag("B"), [t]);
-        assertEquals(em.getTasksWithAnyTag(["A"]), [t]);
-        assertEquals(em.getTasksWithAnyTag(["B"]), [t]);
-        assertEquals(em.getTasksWithAnyTag(["A","B"]), [t]);
-        assertEquals(em.getTasksWithAllTags(["A","B"]), [t]);
-        assertEquals(em.getTasksWithAllTags(["A"]), [t]);
-        assertEquals(em.getTasksWithAllTags(["B"]), [t]);
-    }
-
-    // ENGR-1796: if nothing matched first tag, then returned whatever matched second tag!
-    @Test
-    public void testRetrievingTasksWithAllTagsWhenFirstNotMatched() {
-        Task t = new BasicTask({ /*no-op*/ })
-        em.submit tags:["A"], t
-        t.get();
-
-        assertEquals(em.getTasksWithAllTags(["not_there","A"]), []);
-    }
-    
-    @Test
-    public void testRetrievedTasksIncludesTasksInProgress() {
-        CountDownLatch runningLatch = new CountDownLatch(1);
-        CountDownLatch finishLatch = new CountDownLatch(1);
-        Task t = new BasicTask({ runningLatch.countDown(); finishLatch.await() })
-        em.submit tags:["A"], t
-        
-        try {
-            runningLatch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS);
-    
-            assertEquals(em.getTasksWithTag("A"), [t]);
-        } finally {
-            finishLatch.countDown();
-        }
-    }
-    
-    @Test
-    public void cancelBeforeRun() {
-        CountDownLatch blockForever = new CountDownLatch(1);
-        
-        BasicTask t = [ { blockForever.await(); return 42 } ]
-        t.cancel true
-        assertTrue(t.isCancelled())
-        assertTrue(t.isDone())
-        assertTrue(t.isError())
-        em.submit tag:"A", t
-        try { t.get(); fail("get should have failed due to cancel"); } catch (CancellationException e) {}
-        assertTrue(t.isCancelled())
-        assertTrue(t.isDone())
-        assertTrue(t.isError())
-        
-        log.debug "cancelBeforeRun status: {}", t.getStatusDetail(false)
-        assertTrue(t.getStatusDetail(false).toLowerCase().contains("cancel"))
-    }
-
-    @Test
-    public void cancelDuringRun() {
-        CountDownLatch signalStarted = new CountDownLatch(1);
-        CountDownLatch blockForever = new CountDownLatch(1);
-        
-        BasicTask t = [ { synchronized (data) { signalStarted.countDown(); blockForever.await() }; return 42 } ]
-        em.submit tag:"A", t
-        assertFalse(t.isCancelled())
-        assertFalse(t.isDone())
-        assertFalse(t.isError())
-        
-        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
-        t.cancel true
-        
-        assertTrue(t.isCancelled())
-        assertTrue(t.isError())
-        try { t.get(); fail("get should have failed due to cancel"); } catch (CancellationException e) {}
-        assertTrue(t.isCancelled())
-        assertTrue(t.isDone())
-        assertTrue(t.isError())
-    }
-    
-    @Test
-    public void cancelAfterRun() {
-        BasicTask t = [ { return 42 } ]
-        em.submit tag:"A", t
-
-        assertEquals(42, t.get());
-        t.cancel true
-        assertFalse(t.isCancelled())
-        assertFalse(t.isError())
-        assertTrue(t.isDone())
-    }
-    
-    @Test
-    public void errorDuringRun() {
-        BasicTask t = [ { throw new IllegalStateException("Aaargh"); } ]
-        
-        em.submit tag:"A", t
-        
-        try { t.get(); fail("get should have failed due to error"); } catch (Exception eo) { Exception e = Throwables.getRootCause(eo); assertEquals("Aaargh", e.getMessage()) }
-        
-        assertFalse(t.isCancelled())
-        assertTrue(t.isError())
-        assertTrue(t.isDone())
-        
-        log.debug "errorDuringRun status: {}", t.getStatusDetail(false)
-        assertTrue(t.getStatusDetail(false).contains("Aaargh"), "details="+t.getStatusDetail(false))
-    }
-
-    @Test
-    public void fieldsSetForSimpleBasicTask() {
-        CountDownLatch signalStarted = new CountDownLatch(1);
-        CountDownLatch allowCompletion = new CountDownLatch(1);
-        
-        BasicTask t = [ { signalStarted.countDown(); allowCompletion.await(); return 42 } ]
-        assertEquals(null, t.submittedByTask)
-        assertEquals(-1, t.submitTimeUtc)
-        assertNull(t.getResult())
-
-        em.submit tag:"A", t
-        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
-        
-        assertTrue(t.submitTimeUtc > 0)
-        assertTrue(t.startTimeUtc >= t.submitTimeUtc)
-        assertNotNull(t.getResult())
-        assertEquals(-1, t.endTimeUtc)
-        assertEquals(false, t.isCancelled())
-        
-        allowCompletion.countDown()
-        assertEquals(42, t.get())
-        assertTrue(t.endTimeUtc >= t.startTimeUtc)
-
-        log.debug "BasicTask duration (millis): {}", (t.endTimeUtc - t.submitTimeUtc)
-    }
-
-    @Test
-    public void fieldsSetForBasicTaskSubmittedBasicTask() {
-        //submitted BasicTask B is started by A, and waits for A to complete
-        BasicTask t = new BasicTask( displayName: "sample", description: "some descr", {
-                em.submit tag:"B", {
-                assertEquals(45, em.getTasksWithTag("A").iterator().next().get());
-                46 };
-            45 } )
-        em.submit tag:"A", t
-
-        t.blockUntilEnded()
- 
-//        assertEquals em.getAllTasks().size(), 2
-        
-        BasicTask tb = em.getTasksWithTag("B").iterator().next();
-        assertEquals( 46, tb.get() )
-        assertEquals( t, em.getTasksWithTag("A").iterator().next() )
-        assertNull( t.submittedByTask )
-        
-        BasicTask submitter = tb.submittedByTask;
-        assertNotNull(submitter)
-        assertEquals("sample", submitter.displayName)
-        assertEquals("some descr", submitter.description)
-        assertEquals(t, submitter)
-        
-        assertTrue(submitter.submitTimeUtc <= tb.submitTimeUtc)
-        assertTrue(submitter.endTimeUtc <= tb.endTimeUtc)
-        
-        log.debug "BasicTask {} was submitted by {}", tb, submitter
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.java b/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.java
new file mode 100644
index 0000000..cd0d2b1
--- /dev/null
+++ b/core/src/test/java/brooklyn/util/task/BasicTaskExecutionTest.java
@@ -0,0 +1,460 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package brooklyn.util.task;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.management.Task;
+import brooklyn.test.Asserts;
+import brooklyn.util.collections.MutableMap;
+
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Callables;
+
+/**
+ * Test the operation of the {@link BasicTask} class.
+ *
+ * TODO clarify test purpose
+ */
+public class BasicTaskExecutionTest {
+    private static final Logger log = LoggerFactory.getLogger(BasicTaskExecutionTest.class);
+ 
+    private static final int TIMEOUT_MS = 10*1000;
+    
+    private BasicExecutionManager em;
+    private Map<Object, Object> data;
+
+    @BeforeMethod(alwaysRun=true)
+    public void setUp() {
+        em = new BasicExecutionManager("mycontext");
+        data = Collections.synchronizedMap(new HashMap<Object, Object>());
+        data.clear();
+    }
+    
+    @AfterMethod(alwaysRun=true)
+    public void tearDown() throws Exception {
+        if (em != null) em.shutdownNow();
+        if (data != null) data.clear();
+    }
+    
+    @Test
+    public void runSimpleBasicTask() throws Exception {
+        BasicTask<Object> t = new BasicTask<Object>(newPutCallable(1, "b"));
+        data.put(1, "a");
+        Task<Object> t2 = em.submit(MutableMap.of("tag", "A"), t);
+        assertEquals("a", t.get());
+        assertEquals("a", t2.get());
+        assertEquals("b", data.get(1));
+    }
+    
+    @Test
+    public void runSimpleRunnable() throws Exception {
+        data.put(1, "a");
+        Task<?> t = em.submit(MutableMap.of("tag", "A"), newPutRunnable(1, "b"));
+        assertEquals(null, t.get());
+        assertEquals("b", data.get(1));
+    }
+
+    @Test
+    public void runSimpleCallable() throws Exception {
+        data.put(1, "a");
+        Task<?> t = em.submit(MutableMap.of("tag", "A"), newPutCallable(1, "b"));
+        assertEquals("a", t.get());
+        assertEquals("b", data.get(1));
+    }
+
+    @Test
+    public void runBasicTaskWithWaits() throws Exception {
+        final CountDownLatch signalStarted = new CountDownLatch(1);
+        final CountDownLatch allowCompletion = new CountDownLatch(1);
+        final BasicTask<Object> t = new BasicTask<Object>(new Callable<Object>() {
+            public Object call() throws Exception {
+                Object result = data.put(1, "b");
+                signalStarted.countDown();
+                assertTrue(allowCompletion.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
+                return result;
+            }});
+        data.put(1, "a");
+
+        Task<?> t2 = em.submit(MutableMap.of("tag", "A"), t);
+        assertEquals(t, t2);
+        assertFalse(t.isDone());
+        
+        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
+        assertEquals("b", data.get(1));
+        assertFalse(t.isDone());
+        
+        log.debug("runBasicTaskWithWaits, BasicTask status: {}", t.getStatusDetail(false));
+        
+        Asserts.succeedsEventually(new Runnable() {
+            public void run() {
+                String status = t.getStatusDetail(false);
+                assertTrue(status != null && status.toLowerCase().contains("waiting"), "status="+status);
+            }});
+        
+        allowCompletion.countDown();
+        assertEquals("a", t.get());
+    }
+
+    @Test
+    public void runMultipleBasicTasks() throws Exception {
+        data.put(1, 1);
+        BasicExecutionManager em = new BasicExecutionManager("mycontext");
+        for (int i = 0; i < 2; i++) {
+            em.submit(MutableMap.of("tag", "A"), new BasicTask<Integer>(newIncrementCallable(1)));
+            em.submit(MutableMap.of("tag", "B"), new BasicTask<Integer>(newIncrementCallable((1))));
+        }
+        int total = 0;
+        for (Object tag : em.getTaskTags()) {
+                log.debug("tag {}", tag);
+                for (Task<?> task : em.getTasksWithTag(tag)) {
+                    log.debug("BasicTask {}, has {}", task, task.get());
+                    total += (Integer)task.get();
+                }
+            }
+        assertEquals(10, total);
+        //now that all have completed:
+        assertEquals(5, data.get(1));
+    }
+
+    @Test
+    public void runMultipleBasicTasksMultipleTags() throws Exception {
+        data.put(1, 1);
+        Collection<Task<Integer>> tasks = Lists.newArrayList();
+        tasks.add(em.submit(MutableMap.of("tag", "A"), new BasicTask<Integer>(newIncrementCallable(1))));
+        tasks.add(em.submit(MutableMap.of("tags", ImmutableList.of("A","B")), new BasicTask<Integer>(newIncrementCallable(1))));
+        tasks.add(em.submit(MutableMap.of("tags", ImmutableList.of("B","C")), new BasicTask<Integer>(newIncrementCallable(1))));
+        tasks.add(em.submit(MutableMap.of("tags", ImmutableList.of("D")), new BasicTask<Integer>(newIncrementCallable(1))));
+        int total = 0;
+
+        for (Task<Integer> t : tasks) {
+            log.debug("BasicTask {}, has {}", t, t.get());
+            total += t.get();
+            }
+        assertEquals(10, total);
+ 
+        //now that all have completed:
+        assertEquals(data.get(1), 5);
+        assertEquals(em.getTasksWithTag("A").size(), 2);
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A")).size(), 2);
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A")).size(), 2);
+
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A", "B")).size(), 3);
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A", "B")).size(), 1);
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("B", "C")).size(), 1);
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A", "D")).size(), 3);
+    }
+
+    @Test
+    public void testGetTaskById() throws Exception {
+        Task<?> t = new BasicTask<Void>(newNoop());
+        em.submit(MutableMap.of("tag", "A"), t);
+        assertEquals(em.getTask(t.getId()), t);
+    }
+
+    @Test
+    public void testRetrievingTasksWithTagsReturnsExpectedTask() throws Exception {
+        Task<?> t = new BasicTask<Void>(newNoop());
+        em.submit(MutableMap.of("tag", "A"), t);
+        t.get();
+
+        assertEquals(em.getTasksWithTag("A"), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A", "B")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A")), ImmutableList.of(t));
+    }
+
+    @Test
+    public void testRetrievingTasksWithTagsExcludesNonMatchingTasks() throws Exception {
+        Task<?> t = new BasicTask<Void>(newNoop());
+        em.submit(MutableMap.of("tag", "A"), t);
+        t.get();
+
+        assertEquals(em.getTasksWithTag("B"), ImmutableSet.of());
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("B")), ImmutableSet.of());
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A", "B")), ImmutableSet.of());
+    }
+    
+    @Test
+    public void testRetrievingTasksWithMultipleTags() throws Exception {
+        Task<?> t = new BasicTask<Void>(newNoop());
+        em.submit(MutableMap.of("tags", ImmutableList.of("A", "B")), t);
+        t.get();
+
+        assertEquals(em.getTasksWithTag("A"), ImmutableList.of(t));
+        assertEquals(em.getTasksWithTag("B"), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("B")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAnyTag(ImmutableList.of("A", "B")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A", "B")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("A")), ImmutableList.of(t));
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("B")), ImmutableList.of(t));
+    }
+
+    // ENGR-1796: if nothing matched first tag, then returned whatever matched second tag!
+    @Test
+    public void testRetrievingTasksWithAllTagsWhenFirstNotMatched() throws Exception {
+        Task<?> t = new BasicTask<Void>(newNoop());
+        em.submit(MutableMap.of("tags", ImmutableList.of("A")), t);
+        t.get();
+
+        assertEquals(em.getTasksWithAllTags(ImmutableList.of("not_there","A")), ImmutableSet.of());
+    }
+    
+    @Test
+    public void testRetrievedTasksIncludesTasksInProgress() throws Exception {
+        final CountDownLatch runningLatch = new CountDownLatch(1);
+        final CountDownLatch finishLatch = new CountDownLatch(1);
+        Task<Void> t = new BasicTask<Void>(new Callable<Void>() {
+            public Void call() throws Exception {
+                runningLatch.countDown();
+                finishLatch.await();
+                return null;
+            }});
+        em.submit(MutableMap.of("tags", ImmutableList.of("A")), t);
+        
+        try {
+            runningLatch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    
+            assertEquals(em.getTasksWithTag("A"), ImmutableList.of(t));
+        } finally {
+            finishLatch.countDown();
+        }
+    }
+    
+    @Test
+    public void cancelBeforeRun() throws Exception {
+        final CountDownLatch blockForever = new CountDownLatch(1);
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(new Callable<Integer>() {
+            public Integer call() throws Exception {
+                blockForever.await(); return 42;
+            }});
+        t.cancel(true);
+        assertTrue(t.isCancelled());
+        assertTrue(t.isDone());
+        assertTrue(t.isError());
+        em.submit(MutableMap.of("tag", "A"), t);
+        try {
+            t.get();
+            fail("get should have failed due to cancel");
+        } catch (CancellationException e) {
+            // expected
+        }
+        assertTrue(t.isCancelled());
+        assertTrue(t.isDone());
+        assertTrue(t.isError());
+        
+        log.debug("cancelBeforeRun status: {}", t.getStatusDetail(false));
+        assertTrue(t.getStatusDetail(false).toLowerCase().contains("cancel"));
+    }
+
+    @Test
+    public void cancelDuringRun() throws Exception {
+        final CountDownLatch signalStarted = new CountDownLatch(1);
+        final CountDownLatch blockForever = new CountDownLatch(1);
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(new Callable<Integer>() {
+            public Integer call() throws Exception {
+                synchronized (data) {
+                    signalStarted.countDown();
+                    blockForever.await();
+                }
+                return 42;
+            }});
+        em.submit(MutableMap.of("tag", "A"), t);
+        assertFalse(t.isCancelled());
+        assertFalse(t.isDone());
+        assertFalse(t.isError());
+        
+        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
+        t.cancel(true);
+        
+        assertTrue(t.isCancelled());
+        assertTrue(t.isError());
+        try {
+            t.get();
+            fail("get should have failed due to cancel");
+        } catch (CancellationException e) {
+            // expected
+        }
+        assertTrue(t.isCancelled());
+        assertTrue(t.isDone());
+        assertTrue(t.isError());
+    }
+    
+    @Test
+    public void cancelAfterRun() throws Exception {
+        BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(42));
+        em.submit(MutableMap.of("tag", "A"), t);
+
+        assertEquals(t.get(), (Integer)42);
+        t.cancel(true);
+        assertFalse(t.isCancelled());
+        assertFalse(t.isError());
+        assertTrue(t.isDone());
+    }
+    
+    @Test
+    public void errorDuringRun() throws Exception {
+        BasicTask<Void> t = new BasicTask<Void>(new Callable<Void>() {
+            public Void call() throws Exception {
+                throw new IllegalStateException("Simulating failure in errorDuringRun");
+            }});
+        
+        em.submit(MutableMap.of("tag", "A"), t);
+        
+        try {
+            t.get();
+            fail("get should have failed due to error"); 
+        } catch (Exception eo) { 
+            Throwable e = Throwables.getRootCause(eo);
+            assertEquals("Simulating failure in errorDuringRun", e.getMessage());
+        }
+        
+        assertFalse(t.isCancelled());
+        assertTrue(t.isError());
+        assertTrue(t.isDone());
+        
+        log.debug("errorDuringRun status: {}", t.getStatusDetail(false));
+        assertTrue(t.getStatusDetail(false).contains("Simulating failure in errorDuringRun"), "details="+t.getStatusDetail(false));
+    }
+
+    @Test
+    public void fieldsSetForSimpleBasicTask() throws Exception {
+        final CountDownLatch signalStarted = new CountDownLatch(1);
+        final CountDownLatch allowCompletion = new CountDownLatch(1);
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(new Callable<Integer>() {
+            public Integer call() throws Exception {
+                signalStarted.countDown();
+                allowCompletion.await();
+                return 42;
+            }});
+        assertEquals(null, t.submittedByTask);
+        assertEquals(-1, t.submitTimeUtc);
+        assertNull(t.getResult());
+
+        em.submit(MutableMap.of("tag", "A"), t);
+        assertTrue(signalStarted.await(TIMEOUT_MS, TimeUnit.MILLISECONDS));
+        
+        assertTrue(t.submitTimeUtc > 0);
+        assertTrue(t.startTimeUtc >= t.submitTimeUtc);
+        assertNotNull(t.getResult());
+        assertEquals(-1, t.endTimeUtc);
+        assertEquals(false, t.isCancelled());
+        
+        allowCompletion.countDown();
+        assertEquals(t.get(), (Integer)42);
+        assertTrue(t.endTimeUtc >= t.startTimeUtc);
+
+        log.debug("BasicTask duration (millis): {}", (t.endTimeUtc - t.submitTimeUtc));
+    }
+
+    @Test
+    public void fieldsSetForBasicTaskSubmittedBasicTask() throws Exception {
+        //submitted BasicTask B is started by A, and waits for A to complete
+        BasicTask<Integer> t = new BasicTask<Integer>(MutableMap.of("displayName", "sample", "description", "some descr"), new Callable<Integer>() {
+            public Integer call() throws Exception {
+                em.submit(MutableMap.of("tag", "B"), new Callable<Integer>() {
+                    public Integer call() throws Exception {
+                        assertEquals(45, em.getTasksWithTag("A").iterator().next().get());
+                        return 46;
+                    }});
+                return 45;
+            }});
+        em.submit(MutableMap.of("tag", "A"), t);
+
+        t.blockUntilEnded();
+ 
+//        assertEquals(em.getAllTasks().size(), 2
+        
+        BasicTask<?> tb = (BasicTask<?>) em.getTasksWithTag("B").iterator().next();
+        assertEquals( 46, tb.get() );
+        assertEquals( t, em.getTasksWithTag("A").iterator().next() );
+        assertNull( t.submittedByTask );
+        
+        BasicTask<?> submitter = (BasicTask<?>) tb.submittedByTask;
+        assertNotNull(submitter);
+        assertEquals("sample", submitter.displayName);
+        assertEquals("some descr", submitter.description);
+        assertEquals(t, submitter);
+        
+        assertTrue(submitter.submitTimeUtc <= tb.submitTimeUtc);
+        assertTrue(submitter.endTimeUtc <= tb.endTimeUtc);
+        
+        log.debug("BasicTask {} was submitted by {}", tb, submitter);
+    }
+    
+    private Callable<Object> newPutCallable(final Object key, final Object val) {
+        return new Callable<Object>() {
+            public Object call() {
+                return data.put(key, val);
+            }
+        };
+    }
+    
+    private Callable<Integer> newIncrementCallable(final Object key) {
+        return new Callable<Integer>() {
+            public Integer call() {
+                synchronized (data) {
+                    return (Integer) data.put(key, (Integer)data.get(key) + 1);
+                }
+            }
+        };
+    }
+    
+    private Runnable newPutRunnable(final Object key, final Object val) {
+        return new Runnable() {
+            public void run() {
+                data.put(key, val);
+            }
+        };
+    }
+    
+    private Runnable newNoop() {
+        return new Runnable() {
+            public void run() {
+            }
+        };
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.groovy b/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.groovy
deleted file mode 100644
index 1e4731d..0000000
--- a/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.groovy
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package brooklyn.util.task
-
-import static org.testng.Assert.*
-
-import java.util.concurrent.CopyOnWriteArrayList
-import java.util.concurrent.TimeUnit
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testng.annotations.Test
-
-import brooklyn.test.TestUtils
-import brooklyn.util.internal.TimeExtras
-
-import com.google.common.collect.Lists
-
-public class ScheduledExecutionTest {
-
-	public static final Logger log = LoggerFactory.getLogger(ScheduledExecutionTest.class);
-	
-	static { TimeExtras.init() }
-	
-	@Test
-	public void testScheduledTask() {
-		int PERIOD = 20;
-		BasicExecutionManager m = new BasicExecutionManager();
-		int i=0;
-		def t = new ScheduledTask(delay: 2*PERIOD*TimeUnit.MILLISECONDS, period: PERIOD*TimeUnit.MILLISECONDS, maxIterations: 5, { new BasicTask({
-			log.debug "task running: "+Tasks.current()+" "+Tasks.current().getStatusDetail(false)
-			++i; 
-		}) } );
-	
-		log.info "submitting {} {}", t, t.getStatusDetail(false)
-		m.submit(t);
-		log.info "submitted {} {}", t, t.getStatusDetail(false)
-		int interimResult = t.get()
-		log.info "done one ({}) {} {}", interimResult, t, t.getStatusDetail(false)
-		assertTrue(i>0)
-		t.blockUntilEnded()
-		int finalResult = t.get()
-		log.info "ended ({}) {} {}", finalResult, t, t.getStatusDetail(false)
-		assertEquals(finalResult, 5)
-		assertEquals(i, 5)
-	}
-
-	/** like testScheduledTask but the loop is terminated by the task itself adjusting the period */
-	@Test
-	public void testScheduledTaskSelfEnding() {
-		int PERIOD = 20;
-		BasicExecutionManager m = new BasicExecutionManager();
-		int i=0;
-		def t = new ScheduledTask(delay: 2*PERIOD*TimeUnit.MILLISECONDS, period: PERIOD*TimeUnit.MILLISECONDS, { new BasicTask({
-			if (i>=4) Tasks.current().submittedByTask.period = null
-			log.info "task running (${i}): "+Tasks.current()+" "+Tasks.current().getStatusDetail(false)
-			++i;
-		}) } );
-	
-		log.info "submitting {} {}", t, t.getStatusDetail(false)
-		m.submit(t);
-		log.info "submitted {} {}", t, t.getStatusDetail(false)
-		int interimResult = t.get()
-		log.info "done one ({}) {} {}", interimResult, t, t.getStatusDetail(false)
-		assertTrue(i>0)
-		t.blockUntilEnded()
-		int finalResult = t.get()
-		log.info "ended ({}) {} {}", finalResult, t, t.getStatusDetail(false)
-		assertEquals(finalResult, 5)
-		assertEquals(i, 5)
-	}
-
-	@Test
-	public void testScheduledTaskCancelEnding() {
-		int PERIOD = 20;
-		BasicExecutionManager m = new BasicExecutionManager();
-		int i=0;
-		def t = new ScheduledTask(delay: 2*PERIOD*TimeUnit.MILLISECONDS, period: PERIOD*TimeUnit.MILLISECONDS, { new BasicTask({
-			log.info "task running (${i}): "+Tasks.current()+" "+Tasks.current().getStatusDetail(false)
-			++i;
-			if (i>=5) Tasks.current().submittedByTask.cancel()
-			i
-		}) } );
-	
-		log.info "submitting {} {}", t, t.getStatusDetail(false)
-		m.submit(t);
-		log.info "submitted {} {}", t, t.getStatusDetail(false)
-		int interimResult = t.get()
-		log.info "done one ({}) {} {}", interimResult, t, t.getStatusDetail(false)
-		assertTrue(i>0)
-		t.blockUntilEnded()
-//		int finalResult = t.get()
-		log.info "ended ({}) {} {}", i, t, t.getStatusDetail(false)
-//		assertEquals(finalResult, 5)
-		assertEquals(i, 5)
-	}
-
-    @Test(groups="Integration")
-    public void testScheduledTaskTakesLongerThanPeriod() {
-        final int PERIOD = 1;
-        final int SLEEP_TIME = 100;
-        final int EARLY_RETURN_GRACE = 10;
-        BasicExecutionManager m = new BasicExecutionManager();
-        final List<Long> execTimes = new CopyOnWriteArrayList<Long>();
-        
-        def t = new ScheduledTask(delay: PERIOD*TimeUnit.MILLISECONDS, period: PERIOD*TimeUnit.MILLISECONDS, { new BasicTask({
-            execTimes.add(System.currentTimeMillis());
-            Thread.sleep(100);
-        }) } );
-    
-        m.submit(t);
-        
-        TestUtils.executeUntilSucceeds {
-            execTimes.size() > 3;
-        }
-        
-        List<Long> timeDiffs = Lists.newArrayList();
-        long prevExecTime = -1;
-        for (Long execTime : execTimes) {
-            if (prevExecTime == -1) {
-                prevExecTime = execTime;
-            } else {
-                timeDiffs.add(execTime - prevExecTime);
-                prevExecTime = execTime;
-            }
-        }
-        
-        for (Long timeDiff : timeDiffs) {
-            if (timeDiff < (SLEEP_TIME - EARLY_RETURN_GRACE)) fail("timeDiffs="+timeDiffs+"; execTimes="+execTimes);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.java b/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.java
new file mode 100644
index 0000000..9cda495
--- /dev/null
+++ b/core/src/test/java/brooklyn/util/task/ScheduledExecutionTest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package brooklyn.util.task;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.Test;
+
+import brooklyn.management.Task;
+import brooklyn.test.Asserts;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.exceptions.Exceptions;
+
+import com.google.common.collect.Lists;
+
+public class ScheduledExecutionTest {
+
+	public static final Logger log = LoggerFactory.getLogger(ScheduledExecutionTest.class);
+	
+	@Test
+	public void testScheduledTask() throws Exception {
+		int PERIOD = 20;
+		BasicExecutionManager m = new BasicExecutionManager("mycontextid");
+		final AtomicInteger i = new AtomicInteger(0);
+		ScheduledTask t = new ScheduledTask(MutableMap.of("delay", 2*PERIOD, "period", PERIOD, "maxIterations", 5), new Callable<Task<?>>() {
+            public Task<?> call() throws Exception {
+                return new BasicTask<Integer>(new Callable<Integer>() {
+                    public Integer call() {
+            			log.debug("task running: "+Tasks.current()+" "+Tasks.current().getStatusDetail(false));
+            			return i.incrementAndGet();
+                    }});
+            }});
+	
+		log.info("submitting {} {}", t, t.getStatusDetail(false));
+		m.submit(t);
+		log.info("submitted {} {}", t, t.getStatusDetail(false));
+		Integer interimResult = (Integer) t.get();
+		log.info("done one ({}) {} {}", new Object[] {interimResult, t, t.getStatusDetail(false)});
+		assertTrue(i.get() > 0, "i="+i);
+		t.blockUntilEnded();
+		Integer finalResult = (Integer) t.get();
+		log.info("ended ({}) {} {}", new Object[] {finalResult, t, t.getStatusDetail(false)});
+		assertEquals(finalResult, (Integer)5);
+		assertEquals(i.get(), 5);
+	}
+
+	/** like testScheduledTask but the loop is terminated by the task itself adjusting the period */
+	@Test
+	public void testScheduledTaskSelfEnding() throws Exception {
+		int PERIOD = 20;
+		BasicExecutionManager m = new BasicExecutionManager("mycontextid");
+		final AtomicInteger i = new AtomicInteger(0);
+		ScheduledTask t = new ScheduledTask(MutableMap.of("delay", 2*PERIOD, "period", PERIOD), new Callable<Task<?>>() {
+		    public Task<?> call() throws Exception {
+		        return new BasicTask<Integer>(new Callable<Integer>() {
+		            public Integer call() {
+		                ScheduledTask submitter = (ScheduledTask) ((BasicTask)Tasks.current()).submittedByTask;
+            			if (i.get() >= 4) submitter.period = null;
+            			log.info("task running ("+i+"): "+Tasks.current()+" "+Tasks.current().getStatusDetail(false));
+            			return i.incrementAndGet();
+		            }});
+		    }});
+	
+		log.info("submitting {} {}", t, t.getStatusDetail(false));
+		m.submit(t);
+		log.info("submitted {} {}", t, t.getStatusDetail(false));
+		Integer interimResult = (Integer) t.get();
+		log.info("done one ({}) {} {}", new Object[] {interimResult, t, t.getStatusDetail(false)});
+		assertTrue(i.get() > 0);
+		t.blockUntilEnded();
+		Integer finalResult = (Integer) t.get();
+		log.info("ended ({}) {} {}", new Object[] {finalResult, t, t.getStatusDetail(false)});
+		assertEquals(finalResult, (Integer)5);
+		assertEquals(i.get(), 5);
+	}
+
+	@Test
+	public void testScheduledTaskCancelEnding() throws Exception {
+		int PERIOD = 20;
+		BasicExecutionManager m = new BasicExecutionManager("mycontextid");
+		final AtomicInteger i = new AtomicInteger();
+		ScheduledTask t = new ScheduledTask(MutableMap.of("delay", 2*PERIOD, "period", PERIOD), new Callable<Task<?>>() {
+            public Task<?> call() throws Exception {
+                return new BasicTask<Integer>(new Callable<Integer>() {
+                    public Integer call() {
+            			log.info("task running ("+i+"): "+Tasks.current()+" "+Tasks.current().getStatusDetail(false));
+            			ScheduledTask submitter = (ScheduledTask) ((BasicTask)Tasks.current()).submittedByTask;
+            			i.incrementAndGet();
+            			if (i.get() >= 5) submitter.cancel();
+            			return i.get();
+                    }});
+            }});
+	
+		log.info("submitting {} {}", t, t.getStatusDetail(false));
+		m.submit(t);
+		log.info("submitted {} {}", t, t.getStatusDetail(false));
+		Integer interimResult = (Integer) t.get();
+		log.info("done one ({}) {} {}", new Object[] {interimResult, t, t.getStatusDetail(false)});
+		assertTrue(i.get() > 0);
+		t.blockUntilEnded();
+//		int finalResult = t.get()
+		log.info("ended ({}) {} {}", new Object[] {i, t, t.getStatusDetail(false)});
+//		assertEquals(finalResult, 5)
+		assertEquals(i.get(), 5);
+	}
+
+    @Test(groups="Integration")
+    public void testScheduledTaskTakesLongerThanPeriod() throws Exception {
+        final int PERIOD = 1;
+        final int SLEEP_TIME = 100;
+        final int EARLY_RETURN_GRACE = 10;
+        BasicExecutionManager m = new BasicExecutionManager("mycontextid");
+        final List<Long> execTimes = new CopyOnWriteArrayList<Long>();
+        
+        ScheduledTask t = new ScheduledTask(MutableMap.of("delay", PERIOD, "period", PERIOD), new Callable<Task<?>>() {
+            public Task<?> call() throws Exception {
+                return new BasicTask<Void>(new Runnable() {
+                    public void run() {
+                        execTimes.add(System.currentTimeMillis());
+                        try {
+                            Thread.sleep(100);
+                        } catch (InterruptedException e) {
+                            throw Exceptions.propagate(e);
+                        }
+                    }});
+            }});
+    
+        m.submit(t);
+        
+        Asserts.succeedsEventually(new Runnable() {
+            public void run() {
+                assertTrue(execTimes.size() > 3, "size="+execTimes.size());
+            }});
+        
+        List<Long> timeDiffs = Lists.newArrayList();
+        long prevExecTime = -1;
+        for (Long execTime : execTimes) {
+            if (prevExecTime == -1) {
+                prevExecTime = execTime;
+            } else {
+                timeDiffs.add(execTime - prevExecTime);
+                prevExecTime = execTime;
+            }
+        }
+        
+        for (Long timeDiff : timeDiffs) {
+            if (timeDiff < (SLEEP_TIME - EARLY_RETURN_GRACE)) fail("timeDiffs="+timeDiffs+"; execTimes="+execTimes);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.groovy
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.groovy b/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.groovy
deleted file mode 100644
index debc8e6..0000000
--- a/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.groovy
+++ /dev/null
@@ -1,154 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package brooklyn.util.task
-
-import static brooklyn.test.TestUtils.*
-import static org.testng.Assert.*
-
-import java.util.concurrent.CancellationException
-import java.util.concurrent.CopyOnWriteArrayList
-import java.util.concurrent.CountDownLatch
-import java.util.concurrent.Future
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.TimeoutException
-import java.util.concurrent.atomic.AtomicInteger
-
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testng.annotations.AfterMethod
-import org.testng.annotations.BeforeMethod
-import org.testng.annotations.Test
-
-public class SingleThreadedSchedulerTest {
-
-    private static final Logger log = LoggerFactory.getLogger(SingleThreadedSchedulerTest)
-    
-    private BasicExecutionManager em
-    
-    @BeforeMethod
-    public void setUp() {
-        em = new BasicExecutionManager()
-        em.setTaskSchedulerForTag("category1", SingleThreadedScheduler.class);
-    }
-    
-    @AfterMethod
-    public void tearDown() {
-        em?.shutdownNow()
-    }
-    
-    @Test
-    public void testExecutesInOrder() {
-        final int NUM_TIMES = 1000
-        final List<Integer> result = new CopyOnWriteArrayList()
-        for (i in 0..(NUM_TIMES-1)) {
-            final counter = i
-            em.submit(tag:"category1", { result.add(counter) })
-        }
-        
-        executeUntilSucceeds {
-            assertEquals(result.size(), NUM_TIMES)
-        }
-
-        for (i in 0..(NUM_TIMES-1)) {
-            assertEquals(result.get(i), i)
-        }        
-    }
-    
-    @Test
-    public void testLargeQueueDoesNotConsumeTooManyThreads() {
-        final int NUM_TIMES = 3000
-        final CountDownLatch latch = new CountDownLatch(1)
-        BasicTask blockingTask = [ { latch.await() } ]
-        em.submit tag:"category1", blockingTask
-        
-        final AtomicInteger counter = new AtomicInteger(0)
-        for (i in 1..NUM_TIMES) {
-            BasicTask t = [ {counter.incrementAndGet()} ]
-            em.submit tag:"category1", t
-            if (i % 500 == 0) log.info("Submitted $i jobs...")
-        }
-
-        Thread.sleep(100) // give it more of a chance to create the threads before we let them execute
-        latch.countDown()
-
-        executeUntilSucceeds {
-            assertEquals(counter.get(), NUM_TIMES)
-        }
-    }
-    
-    @Test
-    public void testGetResultOfQueuedTaskBeforeItExecutes() {
-        final CountDownLatch latch = new CountDownLatch(1)
-        em.submit([tag:"category1"], { latch.await() })
-        
-        BasicTask t = [ {return 123} ]
-        Future future = em.submit tag:"category1", t
-
-        new Thread({Thread.sleep(10);latch.countDown()}).start();
-        assertEquals(future.get(), 123)
-    }
-    
-    @Test
-    public void testGetResultOfQueuedTaskBeforeItExecutesWithTimeout() {
-        final CountDownLatch latch = new CountDownLatch(1)
-        em.submit([tag:"category1"], { latch.await() })
-        
-        BasicTask t = [ {return 123} ]
-        Future future = em.submit tag:"category1", t
-
-        try {
-            assertEquals(future.get(10, TimeUnit.MILLISECONDS), 123)
-            fail()
-        } catch (TimeoutException e) {
-            // success
-        }
-    }
-    
-    @Test
-    public void testCancelQueuedTaskBeforeItExecutes() {
-        final CountDownLatch latch = new CountDownLatch(1)
-        em.submit([tag:"category1"], { latch.await() })
-        
-        boolean executed = false
-        BasicTask t = [ {execututed = true} ]
-        Future future = em.submit tag:"category1", t
-
-        future.cancel(true)
-        latch.countDown()
-        Thread.sleep(10)
-        try {
-            future.get()
-        } catch (CancellationException e) {
-            // success
-        }
-        assertFalse(executed)
-    }
-    
-    @Test
-    public void testGetResultOfQueuedTaskAfterItExecutes() {
-        final CountDownLatch latch = new CountDownLatch(1)
-        em.submit([tag:"category1"], { latch.await() })
-        
-        BasicTask t = [ {return 123} ]
-        Future future = em.submit tag:"category1", t
-
-        latch.countDown()
-        assertEquals(future.get(), 123)
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/35363d61/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.java b/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.java
new file mode 100644
index 0000000..265956d
--- /dev/null
+++ b/core/src/test/java/brooklyn/util/task/SingleThreadedSchedulerTest.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package brooklyn.util.task;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.fail;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.test.Asserts;
+import brooklyn.util.collections.MutableMap;
+
+import com.google.common.util.concurrent.Callables;
+
+public class SingleThreadedSchedulerTest {
+
+    private static final Logger log = LoggerFactory.getLogger(SingleThreadedSchedulerTest.class);
+    
+    private BasicExecutionManager em;
+    
+    @BeforeMethod
+    public void setUp() {
+        em = new BasicExecutionManager("mycontextid");
+        em.setTaskSchedulerForTag("category1", SingleThreadedScheduler.class);
+    }
+    
+    @AfterMethod
+    public void tearDown() {
+        if (em != null) em.shutdownNow();
+    }
+    
+    @Test
+    public void testExecutesInOrder() throws Exception {
+        final int NUM_TIMES = 1000;
+        final List<Integer> result = new CopyOnWriteArrayList<Integer>();
+        for (int i = 0; i < NUM_TIMES; i++) {
+            final int counter = i;
+            em.submit(MutableMap.of("tag", "category1"), new Runnable() {
+                public void run() {
+                    result.add(counter);
+                }});
+        }
+        
+        Asserts.succeedsEventually(new Runnable() {
+            @Override public void run() {
+                assertEquals(result.size(), NUM_TIMES);
+            }});
+
+        for (int i = 0; i < NUM_TIMES; i++) {
+            assertEquals(result.get(i), (Integer)i);
+        }        
+    }
+    
+    @Test
+    public void testLargeQueueDoesNotConsumeTooManyThreads() throws Exception {
+        final int NUM_TIMES = 3000;
+        final CountDownLatch latch = new CountDownLatch(1);
+        BasicTask<Void> blockingTask = new BasicTask<Void>(newLatchAwaiter(latch));
+        em.submit(MutableMap.of("tag", "category1"), blockingTask);
+        
+        final AtomicInteger counter = new AtomicInteger(0);
+        for (int i = 0; i < NUM_TIMES; i++) {
+            BasicTask<Void> t = new BasicTask<Void>(new Runnable() {
+                public void run() {
+                    counter.incrementAndGet();
+                }});
+            em.submit(MutableMap.of("tag", "category1"), t);
+            if (i % 500 == 0) log.info("Submitted "+i+" jobs...");
+        }
+
+        Thread.sleep(100); // give it more of a chance to create the threads before we let them execute
+        latch.countDown();
+
+        Asserts.succeedsEventually(new Runnable() {
+            @Override public void run() {
+                assertEquals(counter.get(), NUM_TIMES);
+            }});
+    }
+    
+    @Test
+    public void testGetResultOfQueuedTaskBeforeItExecutes() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch));
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(123));
+        Future<Integer> future = em.submit(MutableMap.of("tag", "category1"), t);
+
+        Thread thread = new Thread(new Runnable() {
+            public void run() {
+                try {
+                    Thread.sleep(10);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+                latch.countDown();
+            }});
+        thread.start();
+        assertEquals(future.get(), (Integer)123);
+    }
+    
+    @Test
+    public void testGetResultOfQueuedTaskBeforeItExecutesWithTimeout() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch));
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(123));
+        Future<Integer> future = em.submit(MutableMap.of("tag", "category1"), t);
+
+        try {
+            assertEquals(future.get(10, TimeUnit.MILLISECONDS), (Integer)123);
+            fail();
+        } catch (TimeoutException e) {
+            // success
+        }
+    }
+    
+    @Test
+    public void testCancelQueuedTaskBeforeItExecutes() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch));
+        
+        final AtomicBoolean executed = new AtomicBoolean();
+        BasicTask<?> t = new BasicTask<Void>(new Runnable() {
+            public void run() {
+                executed.set(true);
+            }});
+        Future<?> future = em.submit(MutableMap.of("tag", "category1"), t);
+
+        future.cancel(true);
+        latch.countDown();
+        Thread.sleep(10);
+        try {
+            future.get();
+        } catch (CancellationException e) {
+            // success
+        }
+        assertFalse(executed.get());
+    }
+    
+    @Test
+    public void testGetResultOfQueuedTaskAfterItExecutes() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch));
+        
+        BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(123));
+        Future<Integer> future = em.submit(MutableMap.of("tag", "category1"), t);
+
+        latch.countDown();
+        assertEquals(future.get(), (Integer)123);
+    }
+    
+    private Callable<Void> newLatchAwaiter(final CountDownLatch latch) {
+        return new Callable<Void>() {
+            public Void call() throws Exception {
+                latch.await();
+                return null;
+            }
+        };
+    }
+}


Mime
View raw message