Make ConfigurableDependentPromise and "onCancel" true decorstors - completes (incl/ cancellation) when delegate is completed, not only when cancel is invoked from the decorator

vsilaev · vsilaev · commit 55bc2b853a10 · 2020-09-20T15:11:33.000+03:00
diff --git a/src/main/java/net/tascalate/concurrent/AggregatingPromise.java b/src/main/java/net/tascalate/concurrent/AggregatingPromise.java
@@ -181,12 +181,13 @@ void onComplete(int idx, T result, Throwable error) {
         }
     }
     
-    void start() {
+    Promise<List<R>> postConstruct() {
         int i = 0;
         for (CompletionStage<? extends T> promise : promises) {
             int idx = i++;
             promise.whenComplete((r, e) -> onComplete(idx, r, e));
         }
+        return this;
     }    
 
     private void markRemainingCancelled() {
diff --git a/src/main/java/net/tascalate/concurrent/ConfigurableDependentPromise.java b/src/main/java/net/tascalate/concurrent/ConfigurableDependentPromise.java
@@ -88,6 +88,28 @@ protected ConfigurableDependentPromise(Promise<T> delegate,
         this.cancellableOrigins = cancellableOrigins;
     }
     
+    private DependentPromise<T> postConstruct() {
+        if (isEmptyArray(cancellableOrigins)) {
+            // Nothing to do
+        }
+        if (isCancelled()) {
+            // Wrapped over already cancelled Promise
+            // So result.cancel() has no effect
+            // and we have to cancel origins explicitly
+            // right after construction
+            cancelPromises(cancellableOrigins, true);
+        } else if (isDone()) {
+            // nothing to do
+        } else {
+            delegate.whenComplete((r, e) -> {
+                if (isCancelled()) {
+                    cancelPromises(cancellableOrigins, true); 
+                }
+            });
+        }        
+        return this;
+    }
+    
     public static <U> DependentPromise<U> from(Promise<U> source) {
         return from(source, PromiseOrigin.NONE);
     }
@@ -103,7 +125,7 @@ protected <U> DependentPromise<U> wrap(Promise<U> original, CompletionStage<?>[]
     private static <U> DependentPromise<U> doWrap(Promise<U> original, 
                                                   Set<PromiseOrigin> defaultEnlistOptions, 
                                                   CompletionStage<?>[] cancellableOrigins) {
-        if (null == cancellableOrigins || cancellableOrigins.length == 0) {
+        if (isEmptyArray(cancellableOrigins)) {
             // Nothing to enlist additionally for this "original" instance
             if (original instanceof ConfigurableDependentPromise) {
                 ConfigurableDependentPromise<U> ioriginal = (ConfigurableDependentPromise<U>)original;
@@ -113,22 +135,15 @@ private static <U> DependentPromise<U> doWrap(Promise<U> original,
                 }
             }
         }
-        ConfigurableDependentPromise<U> result = 
-            new ConfigurableDependentPromise<>(original, defaultEnlistOptions, cancellableOrigins);
         
-        if (result.isCancelled()) {
-            // Wrapped over already cancelled Promise
-            // So result.cancel() has no effect
-            // and we have to cancel origins explicitly
-            // right after construction
-            cancelPromises(result.cancellableOrigins, true);
-        }
-        return result;
+        return new ConfigurableDependentPromise<>(
+            original, defaultEnlistOptions, cancellableOrigins
+        ).postConstruct();
     }
     
     @Override
     public DependentPromise<T> onCancel(Runnable code) {
-        return new ExtraCancellationDependentPromise<>(this, code);
+        return new ExtraCancellationDependentPromise<>(this, code).postConstruct();
     }
     
     // All delay overloads delegate to these methods
@@ -743,12 +758,8 @@ public DependentPromise<T> dependent(Set<PromiseOrigin> defaultEnlistOptions) {
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
-        if (delegate.cancel(mayInterruptIfRunning)) {
-            cancelPromises(cancellableOrigins, mayInterruptIfRunning);
-            return true;
-        } else {
-            return false;
-        }
+        // See postConstruct -- handling is done in delegate.whenComplete
+        return delegate.cancel(mayInterruptIfRunning);
     }
     
     @Override
@@ -864,7 +875,7 @@ private boolean defaultEnlistOrigin() {
     }
 
     static void cancelPromises(CompletionStage<?>[] promises, boolean mayInterruptIfRunning) {
-        if (null != promises) {
+        if (!isEmptyArray(promises)) {
             Arrays.stream(promises)
                   .filter(p -> p != null)
                   .forEach(p -> cancelPromise(p, mayInterruptIfRunning));
@@ -874,6 +885,10 @@ static void cancelPromises(CompletionStage<?>[] promises, boolean mayInterruptIf
     private static boolean identicalSets(Set<?> a, Set<?> b) {
         return a.containsAll(b) && b.containsAll(a);
     }
+    
+    private static boolean isEmptyArray(Object[] array) {
+        return null == array || array.length == 0;
+    }
 
     static class UndecoratedCancellationPromise<T> extends AbstractPromiseDecorator<T, Promise<T>> {
         private final CompletionStage<?>[] dependent;
diff --git a/src/main/java/net/tascalate/concurrent/ExtraCancellationDependentPromise.java b/src/main/java/net/tascalate/concurrent/ExtraCancellationDependentPromise.java
@@ -27,14 +27,19 @@ class ExtraCancellationDependentPromise<T> extends AbstractDependentPromiseDecor
         this.code = code;
     }
     
-    @Override
-    public boolean cancel(boolean mayInterruptIfRunning) {
-        if (super.cancel(mayInterruptIfRunning)) {
+    DependentPromise<T> postConstruct() {
+        if (isCancelled()) {
             code.run();
-            return true;
+        } else if (isDone()) {
+            //
         } else {
-            return false;
+            delegate.whenComplete((r, e) -> {
+               if (isCancelled()) {
+                   code.run();
+               }
+            });
         }
+        return this;
     }
     
     @Override
@@ -44,7 +49,7 @@ protected <U> DependentPromise<U> wrap(CompletionStage<U> original) {
     
     @Override
     public Promise<T> unwrap() {
-        return new Unwrapped<>(delegate, code);
+        return new Unwrapped<>(delegate, code).postConstruct();
     }
     
     @Override
@@ -57,7 +62,7 @@ Promise<T> unwrap(Function<DependentPromise<T>, Promise<T>> fn) {
         if (delegate == unwrapped) {
             return this;
         } else {
-            return new ExtraCancellationPromise.Unwrapped<>(unwrapped, code);
+            return new ExtraCancellationPromise.Unwrapped<>(unwrapped, code).postConstruct();
         }
     }
     
diff --git a/src/main/java/net/tascalate/concurrent/ExtraCancellationPromise.java b/src/main/java/net/tascalate/concurrent/ExtraCancellationPromise.java
@@ -28,14 +28,19 @@ class ExtraCancellationPromise<T> extends AbstractPromiseDecorator<T, Promise<T>
         this.code = code;
     }
     
-    @Override
-    public boolean cancel(boolean mayInterruptIfRunning) {
-        if (super.cancel(mayInterruptIfRunning)) {
+    Promise<T> postConstruct() {
+        if (isCancelled()) {
             code.run();
-            return true;
+        } else if (isDone()) {
+            //
         } else {
-            return false;
+            delegate.whenComplete((r, e) -> {
+               if (isCancelled()) {
+                   code.run();
+               }
+            });
         }
+        return this;
     }
     
     @Override
@@ -45,7 +50,7 @@ protected <U> Promise<U> wrap(CompletionStage<U> original) {
     
     @Override
     public Promise<T> unwrap() {
-        return new Unwrapped<>(delegate, code);
+        return new Unwrapped<>(delegate, code).postConstruct();
     }
     
     @Override
@@ -58,7 +63,7 @@ Promise<T> unwrap(Function<Promise<T>, Promise<T>> fn) {
         if (unwrapped == delegate) {
             return this;
         } else {
-            return new Unwrapped<>(unwrapped, code);
+            return new Unwrapped<>(unwrapped, code).postConstruct();
         }   
     } 
     
diff --git a/src/main/java/net/tascalate/concurrent/Promise.java b/src/main/java/net/tascalate/concurrent/Promise.java
@@ -99,8 +99,9 @@ default boolean isCompletedExceptionally() {
         }
     }
     
+    // @Decorator
     default Promise<T> onCancel(Runnable code) {
-        return new ExtraCancellationPromise<>(this, code);
+        return new ExtraCancellationPromise<>(this, code).postConstruct();
     }
 
     default Promise<T> delay(long timeout, TimeUnit unit) {
@@ -221,6 +222,7 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration,
      * @return
      * created DependentPromise
      */
+    // @Decorator
     default DependentPromise<T> dependent() {
     	return ConfigurableDependentPromise.from(this);
     }
@@ -236,10 +238,12 @@ default DependentPromise<T> dependent() {
      * @return
      * created DependentPromise
      */
+    // @Decorator
     default DependentPromise<T> dependent(Set<PromiseOrigin> defaultEnlistOptions) {
         return ConfigurableDependentPromise.from(this, defaultEnlistOptions);
     }
     
+    // @Decorator
     default Promise<T> defaultAsyncOn(Executor executor) {
         return new ExecutorBoundPromise<>(this, executor);
     }
@@ -258,7 +262,7 @@ default <D> D as(Function<? super Promise<T>, D> decoratorFactory) {
     }
 
     /**
-     * Unwraps underlying {@link Promise} if it was decorated
+     * Unwraps underlying {@link Promise} if it was decorated (removes one level of decorators)
      * @return
      *   the underlying un-decorated {@link Promise} or self if not decorated
      */
@@ -267,7 +271,7 @@ default Promise<T> unwrap() {
     }
     
     /**
-     * Fully unwraps underlying {@link Promise}
+     * Fully unwraps underlying {@link Promise} (removes all decoration layers)
      * @return
      *   the underlying un-decorated {@link Promise} or self if not decorated
      */
diff --git a/src/main/java/net/tascalate/concurrent/Promises.java b/src/main/java/net/tascalate/concurrent/Promises.java
@@ -989,9 +989,8 @@ private static <T, R> Promise<List<R>> atLeast(int minResultsCount, int maxError
             CompletionStage<? extends T> stage = promises.get(0);
             return transform(stage, singleResultMapper, Promises::wrapMultitargetException);
         } else {
-            AggregatingPromise<T, R> result = ctr.create(minResultsCount, maxErrorsCount, cancelRemaining, promises);
-            result.start();
-            return result;
+            return ctr.create(minResultsCount, maxErrorsCount, cancelRemaining, promises)
+                      .postConstruct();
         }
     }
     

Original file line number	Diff line number	Diff line change
`@@ -181,12 +181,13 @@ void onComplete(int idx, T result, Throwable error) {`
`181`	`181`	`}`
`182`	`182`	`}`
`183`	`183`
`184`		`- void start() {`
	`184`	`+ Promise<List<R>> postConstruct() {`
`185`	`185`	`int i = 0;`
`186`	`186`	`for (CompletionStage<? extends T> promise : promises) {`
`187`	`187`	`int idx = i++;`
`188`	`188`	`promise.whenComplete((r, e) -> onComplete(idx, r, e));`
`189`	`189`	`}`
	`190`	`+ return this;`
`190`	`191`	`}`
`191`	`192`
`192`	`193`	`private void markRemainingCancelled() {`
Original file line number	Diff line number	Diff line change
`@@ -99,8 +99,9 @@ default boolean isCompletedExceptionally() {`
`99`	`99`	`}`
`100`	`100`	`}`
`101`	`101`
	`102`	`+ // @Decorator`
`102`	`103`	`default Promise<T> onCancel(Runnable code) {`
`103`		`- return new ExtraCancellationPromise<>(this, code);`
	`104`	`+ return new ExtraCancellationPromise<>(this, code).postConstruct();`
`104`	`105`	`}`
`105`	`106`
`106`	`107`	`default Promise<T> delay(long timeout, TimeUnit unit) {`
`@@ -221,6 +222,7 @@ default Promise<T> onTimeout(Supplier<? extends T> supplier, Duration duration,`
`221`	`222`	`* @return`
`222`	`223`	`* created DependentPromise`
`223`	`224`	`*/`
	`225`	`+ // @Decorator`
`224`	`226`	`default DependentPromise<T> dependent() {`
`225`	`227`	`return ConfigurableDependentPromise.from(this);`
`226`	`228`	`}`
`@@ -236,10 +238,12 @@ default DependentPromise<T> dependent() {`
`236`	`238`	`* @return`
`237`	`239`	`* created DependentPromise`
`238`	`240`	`*/`
	`241`	`+ // @Decorator`
`239`	`242`	`default DependentPromise<T> dependent(Set<PromiseOrigin> defaultEnlistOptions) {`
`240`	`243`	`return ConfigurableDependentPromise.from(this, defaultEnlistOptions);`
`241`	`244`	`}`
`242`	`245`
	`246`	`+ // @Decorator`
`243`	`247`	`default Promise<T> defaultAsyncOn(Executor executor) {`
`244`	`248`	`return new ExecutorBoundPromise<>(this, executor);`
`245`	`249`	`}`
`@@ -258,7 +262,7 @@ default <D> D as(Function<? super Promise<T>, D> decoratorFactory) {`
`258`	`262`	`}`
`259`	`263`
`260`	`264`	`/**`
`261`		`- * Unwraps underlying {@link Promise} if it was decorated`
	`265`	`+ * Unwraps underlying {@link Promise} if it was decorated (removes one level of decorators)`
`262`	`266`	`* @return`
`263`	`267`	`* the underlying un-decorated {@link Promise} or self if not decorated`
`264`	`268`	`*/`
`@@ -267,7 +271,7 @@ default Promise<T> unwrap() {`
`267`	`271`	`}`
`268`	`272`
`269`	`273`	`/**`
`270`		`- * Fully unwraps underlying {@link Promise}`
	`274`	`+ * Fully unwraps underlying {@link Promise} (removes all decoration layers)`
`271`	`275`	`* @return`
`272`	`276`	`* the underlying un-decorated {@link Promise} or self if not decorated`
`273`	`277`	`*/`
Original file line number	Diff line number	Diff line change
`@@ -989,9 +989,8 @@ private static <T, R> Promise<List<R>> atLeast(int minResultsCount, int maxError`
`989`	`989`	`CompletionStage<? extends T> stage = promises.get(0);`
`990`	`990`	`return transform(stage, singleResultMapper, Promises::wrapMultitargetException);`
`991`	`991`	`} else {`
`992`		`- AggregatingPromise<T, R> result = ctr.create(minResultsCount, maxErrorsCount, cancelRemaining, promises);`
`993`		`- result.start();`
`994`		`- return result;`
	`992`	`+ return ctr.create(minResultsCount, maxErrorsCount, cancelRemaining, promises)`
	`993`	`+ .postConstruct();`
`995`	`994`	`}`
`996`	`995`	`}`
`997`	`996`