Fix issue #6 -- AbstractCompletableTask.thenComposeAsync does not propagate exception

Author: vsilaev

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

@@ -248,40 +248,49 @@ public <U> Promise<U> thenComposeAsync(Function<? super T, ? extends CompletionS
         addCallbacks(
             tempStage, 
             consumerAsFunction(r -> {
-                CompletionStage<U> returned = fn.apply(r);
-                // tempStage is completed successfully, so no sense
-                // to include it in cancellableOrigins
-                // However, nextStage is in progress
-                // IMPORTANT: it COULD be shared, but typically is not
-                // So in very rare case some nasty behavior MAY exist 
-                // if others depends on it
-                
-                // TEST: There is a race when fn.apply(r) is completed
-                // normally and nextStage is cancelled before returned is set
-                // as its nextStage's cancellableOrigins. In this case,
-                // execution of returned continues as nextStage cannot be
-                // cancelled for a second time. However, completion stages after
-                // nextStage are completed exceptionally (correctly) since when
-                // moveToNextStage is executed nextStage is already completed
-                // (cancelled) from cancel(...) -> onError(...). In order to
-                // cancel returned here, I think you need to know whether
-                // nextStage might have been interrupted.
-                // try {
-                //    Thread.sleep(100);
-                //} catch (InterruptedException ex) {
-                //}
-                
-                nextStage.resetCancellableOrigins(returned);
-                if (nextStage.isCancelled()) {
-                    nextStage.cancelOrigins(true);
-                } else {
-                    returned.whenComplete(moveToNextStage);
+                try {
+                    CompletionStage<U> returned = fn.apply(r);
+                    // tempStage is completed successfully, so no sense
+                    // to include it in cancellableOrigins
+                    // However, nextStage is in progress
+                    // IMPORTANT: it COULD be shared, but typically is not
+                    // So in very rare case some nasty behavior MAY exist 
+                    // if others depends on it
+                    
+                    // TEST: There is a race when fn.apply(r) is completed
+                    // normally and nextStage is cancelled before returned is set
+                    // as its nextStage's cancellableOrigins. In this case,
+                    // execution of returned continues as nextStage cannot be
+                    // cancelled for a second time. However, completion stages after
+                    // nextStage are completed exceptionally (correctly) since when
+                    // moveToNextStage is executed nextStage is already completed
+                    // (cancelled) from cancel(...) -> onError(...). In order to
+                    // cancel returned here, I think you need to know whether
+                    // nextStage might have been interrupted.
+                    // try {
+                    //    Thread.sleep(100);
+                    //} catch (InterruptedException ex) {
+                    //}
+                    
+                    nextStage.resetCancellableOrigins(returned);
+                    if (nextStage.isCancelled()) {
+                        nextStage.cancelOrigins(true);
+                    } else {
+                        returned.whenComplete(moveToNextStage);
+                    }
+                } catch (Throwable ex) {
+                    // must-have if fn.apply above failed 
+                    nextStage.resetCancellableOrigins((CompletionStage<U>)null);
+                    // no need to check nextStage.isCancelled()
+                    // while there are no origins to cancel
+                    // propagate error immediately 
+                    moveToNextStage.accept(null, ex);  
                 }
             }), 
             e -> { 
                 moveToNextStage.accept(null, e); 
                 return null; 
-            }, /* must-have if fn.apply above failed */
+            },
             executor
         );
 

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

@@ -15,6 +15,8 @@
  */
 package net.tascalate.concurrent;
 
+import static org.junit.Assert.fail;
+
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.After;
 import org.junit.Assert;
@@ -76,6 +78,23 @@ public void testThenComposeAsyncRace() {
         Assert.assertTrue("Expected cancelled true, but is " + cancelled.get(), cancelled.get());
     }
 
+    
+    @Test
+    public void testThenComposeException() {
+        Promise<Void> p = CompletableTask.runAsync(() -> trySleep(10), executor)
+                .thenCompose(it -> {
+                    throw new IllegalStateException("oh no!");
+                });
+        try {
+            p.get();
+        } catch (Exception ex) {
+            // expected
+            return;
+        }
+        fail("Exception must be thrown");
+    }
+    
+    
     private void trySleep(long millis) {
         try {
             Thread.sleep(millis);