Simplify timeouts cleanup

Author: vsilaev

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

@@ -173,16 +173,10 @@ public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout)
     public DependentPromise<T> orTimeout(Duration duration, boolean cancelOnTimeout, boolean enlistOrigin) {
         Promise<T> onTimeout = Promises.failAfter(duration);
         // Use *async to execute on default "this" executor
-        return this
-            .applyToEitherAsync(onTimeout, Function.identity(), enlistOrigin ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY)
-            .whenComplete((v, e) -> {
-        	    // Result comes from timeout and cancel-on-timeout is set
-        	    // If both are done then cancel has no effect anyway
-                if (onTimeout.isDone() && cancelOnTimeout) {
-                    cancel(true);
-                }
-                onTimeout.cancel(true); 
-            }, true);        
+        DependentPromise<T> result = this
+            .applyToEitherAsync(onTimeout, Function.identity(), enlistOrigin ? PromiseOrigin.ALL : PromiseOrigin.PARAM_ONLY);
+        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        return result;
     }
 
     public DependentPromise<T> onTimeout(T value, long timeout, TimeUnit unit) {
@@ -237,19 +231,14 @@ public DependentPromise<T> onTimeout(Supplier<T> supplier, Duration duration, bo
             .dependent(Promises.delay(duration))
             .thenApply(d -> supplier, true);
 
-        return this
+        DependentPromise<T> result = this
             // resolved value converted to supplier
             .thenApply(valueToSupplier, enlistOrigin)
             // Use *async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, Supplier::get,  PromiseOrigin.ALL)
-            .whenComplete((v, e) -> {
-            	// Result comes from timeout and cancel-on-timeout is set
-            	// If both are done then cancel has no effect anyway
-                if (onTimeout.isDone() && cancelOnTimeout) {
-                    cancel(true);
-                }
-                onTimeout.cancel(true);                 
-            }, true);        
+            .applyToEitherAsync(onTimeout, Supplier::get,  PromiseOrigin.ALL);
+        
+        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        return result;
     }
 
     public <U> DependentPromise<U> thenApply(Function<? super T, ? extends U> fn) {

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

@@ -75,16 +75,11 @@ default Promise<T> orTimeout(Duration duration) {
     default Promise<T> orTimeout(Duration duration, boolean cancelOnTimeout) {
         Promise<T> onTimeout = Promises.failAfter(duration);
         // Use *async to execute on default "this" executor
-        return Promises.dependent(this)
-            .applyToEitherAsync(onTimeout, Function.identity(), PromiseOrigin.PARAM_ONLY)
-            .whenComplete((v, e) -> {
-            	// Result comes from timeout and cancel-on-timeout is set
-            	// If both are done then cancel has no effect anyway
-                if (onTimeout.isDone() && cancelOnTimeout) {
-                    cancel(true);
-                }
-                onTimeout.cancel(true); 
-            }, true);
+        Promise<T> result = Promises.dependent(this)
+            .applyToEitherAsync(onTimeout, Function.identity(), PromiseOrigin.PARAM_ONLY);
+        
+        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        return result;
     }
 
     default Promise<T> onTimeout(T value, long timeout, TimeUnit unit) {
@@ -123,19 +118,14 @@ default Promise<T> onTimeout(Supplier<T> supplier, Duration duration, boolean ca
             .dependent(Promises.delay(duration))
             .thenApply(d -> supplier, true);
 
-        return Promises.dependent(this)
+        Promise<T> result = Promises.dependent(this)
             // resolved value converted to supplier
             .thenApply(valueToSupplier, false)
             // Use *async to execute on default "this" executor
-            .applyToEitherAsync(onTimeout, Supplier::get, PromiseOrigin.ALL)
-            .whenComplete((v, e) -> {
-            	// Result comes from timeout and cancel-on-timeout is set
-            	// If both are done then cancel has no effect anyway
-                if (onTimeout.isDone() && cancelOnTimeout) {
-                    cancel(true);
-                }
-                onTimeout.cancel(true);
-            }, true);
+            .applyToEitherAsync(onTimeout, Supplier::get, PromiseOrigin.ALL);
+        
+        result.whenComplete(Promises.timeoutsCleanup(this, onTimeout, cancelOnTimeout));
+        return result;
     }
 
     public <U> Promise<U> thenApply(Function<? super T, ? extends U> fn);

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

@@ -657,6 +657,17 @@ static Duration toDuration(long delay, TimeUnit timeUnit) {
         return Duration.of(delay, toChronoUnit(timeUnit));
     }
 
+    static <T, U> BiConsumer<T, U> timeoutsCleanup(Promise<T> self, Promise<?> onTimeout, boolean cancelOnTimeout) {
+    	return (r, e) -> {
+        	// Result comes from timeout and cancel-on-timeout is set
+        	// If both are done then cancel has no effect anyway
+            if ((onTimeout.isDone() && !onTimeout.isCancelled()) && cancelOnTimeout) {
+                self.cancel(true);
+            }
+            onTimeout.cancel(true);
+    	};
+    }
+    
     private static ChronoUnit toChronoUnit(TimeUnit unit) { 
         Objects.requireNonNull(unit, "unit"); 
         switch (unit) {