adding unit operators to CompletableTask; adding some tests

Author: vsilaev

pom.xml

@@ -4,7 +4,7 @@
 
 	<groupId>net.tascalate.concurrent</groupId>
 	<artifactId>net.tascalate.concurrent.lib</artifactId>
-	<version>1.0-SNAPSHOT</version>
+	<version>0.5-SNAPSHOT</version>
 	<packaging>jar</packaging>
 
 	<name>Tascalate Concurrenct</name>
@@ -53,6 +53,7 @@
 		</dependencies>
 	</dependencyManagement>
 
+
 	<distributionManagement>
 		<snapshotRepository>
 			<id>ossrh</id>
@@ -115,6 +116,19 @@
        </profiles>
 
 
+	<dependencies>
+		<dependency>
+			<groupId>commons-logging</groupId>
+			<artifactId>commons-logging</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
 	<build>
 		<plugins>
 			<!--

src/main/java/net/tascalate/concurrent/CompletableTask.java

@@ -21,7 +21,7 @@
 
 public class CompletableTask<T> extends AbstractCompletableTask<T> implements RunnableFuture<T> {
 
-    public CompletableTask(final Executor executor, Callable<T> callable) {
+    protected CompletableTask(final Executor executor, Callable<T> callable) {
         super(executor, callable);
     }
 
@@ -30,11 +30,15 @@ public void run() {
         task.run();
     }
 
-    public static <T> Promise<T> completedFuture(T value, Executor defaultExecutor) {
+    public static <T> Promise<T> resolve(T value, Executor defaultExecutor) {
         CompletableTask<T> result = new CompletableTask<T>(defaultExecutor, () -> value);
         SAME_THREAD_EXECUTOR.execute(result);
         return result;
     }
+    
+    public static Promise<Void> asyncOn(Executor defaultExecutor) {
+        return resolve(null, defaultExecutor);
+    }
 
     @Override
     Runnable setupTransition(Callable<T> code) {

src/test/java/net/tascalate/concurrent/EitherCompletableTaskTests.java

@@ -0,0 +1,78 @@
+package net.tascalate.concurrent;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class EitherCompletableTaskTests {
+    
+    private TaskExecutorService executor;
+
+    @Before
+    public void setup() {
+        executor = TaskExecutors.newFixedThreadPool(4);
+    }
+    
+    @After
+    public void tearDown() {
+        executor.shutdown();
+    }
+    
+    @Test
+    public void testAnyTaskIsExecuted1() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null, null};
+        String[] results = {null, null};
+        AtomicInteger idx = new AtomicInteger(0);
+
+        Promise<String> first = executor.submit(() -> resultWithDelay("ABC", 2000)); 
+        Promise<String> second = executor.submit(() -> resultWithDelay("XYZ", 100)); // Second wins -- shorter delay
+
+        first
+        .applyToEitherAsync(second, String::toLowerCase)
+        .whenComplete((r, e) -> {
+            int i = idx.getAndIncrement();
+            results[i] = r;
+            errors[i]  = e;
+            ready.countDown();
+        });
+        
+        ready.await();
+        Assert.assertNull(errors[0]);
+        Assert.assertEquals(results[0], "xyz");
+    }
+    
+    @Test
+    public void testAnyTaskIsExecuted2() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null, null};
+        String[] results = {null, null};
+        AtomicInteger idx = new AtomicInteger(0);
+
+        Promise<String> first = executor.submit(() -> resultWithDelay("ABC", 2000)); 
+        Promise<String> second = executor.submit(() -> resultWithDelay("XYZ", 100)); // Second wins -- shorter delay
+
+        second
+        .applyToEitherAsync(first, String::toLowerCase)
+        .whenComplete((r, e) -> {
+            int i = idx.getAndIncrement();
+            results[i] = r;
+            errors[i]  = e;
+            ready.countDown();
+        });
+        
+        ready.await();
+        Assert.assertNull(errors[0]);
+        Assert.assertEquals(results[0], "xyz");
+    }
+
+    
+    <T> T resultWithDelay(T value, long delay) throws InterruptedException {
+        Thread.sleep(delay);
+        return value;
+    }
+}

src/test/java/net/tascalate/concurrent/SimpleCompletableTaskTests.java

@@ -0,0 +1,97 @@
+package net.tascalate.concurrent;
+
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class SimpleCompletableTaskTests {
+    
+    private TaskExecutorService executor;
+
+    @Before
+    public void setup() {
+        executor = TaskExecutors.newFixedThreadPool(4);
+    }
+    
+    @After
+    public void tearDown() {
+        executor.shutdown();
+    }
+    
+    @Test
+    public void testTaskIsExecuted1() throws Exception {
+        Promise<Integer> promise = executor.submit(() -> 50);
+        int i = promise.get();
+        Assert.assertEquals(i, 50);
+    }
+    
+    @Test
+    public void testTaskIsExecuted2() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null};
+        String[] results = {null};
+        executor.submit(() -> "ABC").whenComplete((r, e) -> {
+            results[0] = r;
+            errors[0]  = e;
+            ready.countDown();
+        });
+        ready.await();
+        Assert.assertNull(errors[0]);
+        Assert.assertEquals(results[0], "ABC");
+    }
+    
+    @Test
+    public void testTaskIsFailed() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null};
+        Integer[] results = {null};
+        executor.submit(this::failedResult).whenComplete((r, e) -> {
+            results[0] = r;
+            errors[0]  = e;
+            ready.countDown();
+        });
+        ready.await();
+        Assert.assertNull(results[0]);
+        Assert.assertTrue("ArithmeticException was not raised", errors[0] instanceof ArithmeticException);
+    }
+    
+    @Test
+    public void testNestedTaskIsExecuted() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null};
+        Integer[] results = {null};
+        executor.submit(() -> 50).thenApply(i -> i * 10).whenComplete((r, e) -> {
+            results[0] = r;
+            errors[0]  = e;
+            ready.countDown();
+        });
+        ready.await();
+        Assert.assertNull(errors[0]);
+        Assert.assertEquals(results[0], Integer.valueOf(500));
+    }
+    
+    @Test
+    public void testNestedTaskIsFailed() throws Exception {
+        CountDownLatch ready = new CountDownLatch(1);
+        Throwable[] errors = {null};
+        Integer[] results = {null};
+        executor.submit(this::failedResult).thenApply(i -> i * 10).whenComplete((r, e) -> {
+            results[0] = r;
+            errors[0]  = e;
+            ready.countDown();
+        });
+        ready.await();
+        Assert.assertNull(results[0]);
+        Assert.assertTrue("CompletionException was not raised", errors[0] instanceof CompletionException);
+        Assert.assertTrue("CompletionException is not caused by ArithmeticException", errors[0].getCause() instanceof ArithmeticException);        
+    }
+
+    
+    int failedResult() {
+        throw new ArithmeticException();
+    }
+}