Standardize CompletableFutureWrapper.toCompletableFuture(enlistOrigin) behavior

Author: vsilaev

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

@@ -54,6 +54,18 @@ boolean complete(T value, Throwable ex) {
         return null == ex ? success(value) : failure(ex);
     }
 
+    @Override
+    public CompletableFuture<T> toCompletableFuture() {
+        // Return concrete subclass that neither completes nor cancels this wrapper
+        return (CompletableFuture<T>)delegate.thenApply(Function.identity());
+    }
+    
+    // Report self-origin
+    @Override
+    public CompletionStage<T> τ() {
+        return this;
+    }
+    
     @Override
     protected <U> Promise<U> wrap(CompletionStage<U> original) {
         return new CompletableFutureWrapper<>((CompletableFuture<U>)original);

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

@@ -112,6 +112,17 @@ public CompletionStage<T> minimalCompletionStage() {
         // Should pass this, not delegate - while delegate has invalid thenCompose impl.
         return new MinimalCompletionStage<>(this);
     }
+    
+    @Override
+    public CompletableFuture<T> toCompletableFuture() {
+        // Returns the delegate itself, so this CompletablePromise may be completed / cancelled
+        return delegate;
+    }
+    
+    @Override
+    public CompletionStage<T> τ() {
+        return delegate;
+    }
 
     @Override
     protected <U> CompletablePromise<U> wrap(CompletionStage<U> original) {

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

@@ -36,6 +36,7 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
+import net.tascalate.concurrent.core.Delegator;
 import net.tascalate.concurrent.decorators.AbstractPromiseDecorator; 
 
 /**
@@ -74,7 +75,7 @@
  * @param <T>
  *   a type of the successfully resolved promise value    
  */
-public class ConfigurableDependentPromise<T> implements DependentPromise<T> {
+public class ConfigurableDependentPromise<T> implements DependentPromise<T>, Delegator<T> {
     protected final Promise<T> delegate;
     protected final CompletionStage<?>[] cancellableOrigins;
     protected final Set<PromiseOrigin> defaultEnlistOptions;
@@ -830,21 +831,61 @@ protected Promise<T> cancellablePromiseOf(Promise<T> original) {
     public CompletableFuture<T> toCompletableFuture() {
         return toCompletableFuture(defaultEnlistOrigin());
     }
-    
+
     @Override
     public CompletableFuture<T> toCompletableFuture(boolean enlistOrigin) {
-        if (!enlistOrigin) {
-            return delegate.toCompletableFuture();
-        } else {
-            CompletableFuture<T> result = new CompletableFuture<T>() {
+        CompletionStage<T> tau = tauOf(delegate);
+        CompletableFuture<T> rootDelegate = tau instanceof CompletableFuture ? (CompletableFuture<T>)tau : null; 
+        CompletableFuture<T> result;
+        if (enlistOrigin) {
+            if (rootDelegate != null) {
+                // Use toCompletableFuture only if there is a delegate
+                CompletableFuture<T> defaultResult = delegate.toCompletableFuture();
+                if (rootDelegate == defaultResult) {
+                    return defaultResult;
+                }
+            }
+            result = new CompletableFuture<T>() {
                 @Override 
                 public boolean cancel(boolean mayInterruptIfRunning) {
                     ConfigurableDependentPromise.this.cancel(mayInterruptIfRunning);
                     return super.cancel(mayInterruptIfRunning);
                 }
             };
-            whenComplete((r, e) -> iif(null == e ? result.complete(r) : result.completeExceptionally(e)));
-            return result;
+        } else {
+            CompletableFuture<T> defaultResult = delegate.toCompletableFuture();
+            if (rootDelegate != defaultResult) {
+                // Assume if there is no rootDelegate or rootDelegate != rootDelegate.toCompletableFuture
+                // then toCompletableFuture returns stage that doesn't cancel original one
+                return defaultResult;
+            }
+            result = new CompletableFuture<>();
+        }
+        whenComplete((r, e) -> iif(null == e ? result.complete(r) : result.completeExceptionally(e)));
+        return result;
+    }
+    
+    @Override
+    public CompletionStage<T> τ() {
+        return tauOf(delegate);
+    }
+    
+    @Override
+    public String toString() {
+        return String.format(
+            "%s@%d[delegate=%s, enlistOptions=%s, cancellable=%s]", 
+            getClass().getSimpleName(), System.identityHashCode(this), 
+            delegate, this.defaultEnlistOptions, null == cancellableOrigins ? "{}" : Arrays.asList(cancellableOrigins)
+        );
+    }
+    
+    private static <T> CompletionStage<T> tauOf(Promise<T> promise) {
+        if (promise instanceof Delegator) {
+            @SuppressWarnings("unchecked")
+            Delegator<T> delegator = (Delegator<T>)promise;
+            return delegator.τ();
+        } else {
+            return promise;
         }
     }
 

src/main/java/net/tascalate/concurrent/core/Delegator.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright 2015-2020 Valery Silaev (http://vsilaev.com)
+ *
+ * Licensed 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 net.tascalate.concurrent.core;
+
+import java.util.concurrent.CompletionStage;
+
+public interface Delegator<T> {
+    CompletionStage<T> τ();
+}

src/main/java/net/tascalate/concurrent/decorators/AbstractPromiseLikeDecorator.java

@@ -23,10 +23,11 @@
 import java.util.function.Function;
 
 import net.tascalate.concurrent.Promise;
+import net.tascalate.concurrent.core.Delegator;
 
 public abstract class AbstractPromiseLikeDecorator<T, D extends CompletionStage<T>>
     extends AbstractCompletionStageDecorator<T, D>
-    implements CompletionStage<T> {
+    implements CompletionStage<T>, Delegator<T> {
 
     protected AbstractPromiseLikeDecorator(D delegate) {
         super(delegate);
@@ -270,4 +271,29 @@ public <U> Promise<U> handleAsync(BiFunction<? super T, Throwable, ? extends U>
     public <U> Promise<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
         return cast(super.handleAsync(fn, executor));
     }
+    
+    // Tau-operator -- the origin of origins
+    public CompletionStage<T> τ() {
+        CompletionStage<T> p = delegate;
+        if (p instanceof Delegator) {
+            return delegator(p).τ();
+        } else {
+            // Default path -- unroll
+            while (p instanceof AbstractCompletionStageDecorator) {
+                @SuppressWarnings("unchecked")
+                AbstractCompletionStageDecorator<T, ? extends CompletionStage<T>> ap = 
+                    (AbstractCompletionStageDecorator<T, ? extends CompletionStage<T>>)p;
+                p = ap.delegate;
+                if (p instanceof Delegator) {
+                    return delegator(p).τ();
+                }
+            }
+            return p;
+        }
+    }
+    
+    @SuppressWarnings("unchecked")
+    private static <T> Delegator<T> delegator(CompletionStage<T> delegate) {
+        return (Delegator<T>)delegate;
+    }
 }

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

@@ -32,6 +32,7 @@
 
 import net.tascalate.concurrent.core.CompletionStageAPI;
 import net.tascalate.concurrent.decorators.AbstractCompletionStageDecorator;
+import net.tascalate.concurrent.decorators.CustomizablePromiseDecorator;
 import net.tascalate.concurrent.decorators.ExtendedPromiseDecorator;
 import net.tascalate.concurrent.locks.AsyncSemaphoreLock;
 
@@ -106,6 +107,10 @@ public static void main(final String[] argv) throws InterruptedException, Execut
 
         final TaskExecutorService executorService = TaskExecutors.newFixedThreadPool(6, tf);
 
+        System.out.println(new CustomizablePromiseDecorator<>(CompletableTask.completed("", executorService).defaultAsyncOn(executorService), null).τ());
+        System.out.println(new CustomizablePromiseDecorator<>(Promises.success(11).defaultAsyncOn(executorService), null).τ());
+        System.out.println(new CustomizablePromiseDecorator<>(new CompletablePromise<Long>().defaultAsyncOn(executorService), null).τ());
+        
         Promise<?> t = 
         Promises.success("11")
         //CompletableTask