Adding bi-completion to CompletableFutureWrapper / CompletablePromise; using it in impls.

Author: vsilaev

pom.xml

@@ -173,6 +173,9 @@
 				<configuration>
 					<source>1.8</source>
 					<target>1.8</target>
+					<!--
+					<debug>false</debug>
+					-->
 				</configuration>
 			</plugin>
 			<plugin>

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

@@ -51,6 +51,10 @@ protected boolean onFailure(Throwable ex) {
         return delegate.completeExceptionally(ex);
     }    
 
+    boolean complete(T value, Throwable ex) {
+        return null == ex ? success(value) : failure(ex);
+    }
+    
     @Override
     protected <U> Promise<U> wrap(CompletionStage<U> original) {
         return new CompletableFutureWrapper<>((CompletableFuture<U>)original);

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

@@ -48,10 +48,15 @@ public boolean complete(T value) {
         return success(value);
     }
 
-    public boolean  completeExceptionally(Throwable ex) {
+    public boolean completeExceptionally(Throwable ex) {
         return failure(ex);
     }
 
+    @Override
+    public boolean complete(T value, Throwable ex) {
+        return super.complete(value, ex);
+    }
+    
     public Promise<T> completeAsync(Supplier<? extends T> supplier) {
         CompletionStageAPI api = CompletionStageAPI.current();
         Promise<T> result = completeAsync(supplier, api.defaultExecutorOf(delegate));
@@ -82,7 +87,7 @@ public CompletablePromise<T> copy() {
             }
             return result;
         } else {
-            whenComplete((r, e) -> iif(null == e ? result.complete(r) : result.completeExceptionally(e)));
+            whenComplete(result::complete);
             return result;
         }
     }
@@ -97,7 +102,7 @@ public Promise<T> minimalPromise() {
             }
             return result;            
         } else {
-            whenComplete((r, e) -> iif(null == e ? result.success(r) : result.failure(e)));
+            whenComplete(result::complete);
             return result;
         }
     }

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

@@ -18,7 +18,6 @@
 import static net.tascalate.concurrent.SharedFunctions.NO_SUCH_ELEMENT;
 import static net.tascalate.concurrent.SharedFunctions.cancelPromise;
 import static net.tascalate.concurrent.SharedFunctions.failure;
-import static net.tascalate.concurrent.SharedFunctions.iif;
 import static net.tascalate.concurrent.SharedFunctions.selectFirst;
 
 import java.time.Duration;
@@ -908,8 +907,8 @@ public boolean cancel(boolean mayInterruptIfRunning) {
                     }
                 }
             };
-            whenComplete((r, e) -> iif(null == e ? result.success(r) : result.failure(e)));
-            return result.toCompletableFuture();
+            whenComplete(result::complete);
+            return result.toCompletableFuture(); // Effectively result.delegate
         }
     }
 

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

@@ -126,7 +126,6 @@ public static <T> Promise<T> loop(T initialValue,
         asyncLoop.run(initialValue);
         return asyncLoop;  
     }
-
 
     public static <T, R extends AutoCloseable> Promise<T> tryApply(CompletionStage<R> resourcePromise,
                                                                    Function<? super R, ? extends T> fn) {
@@ -148,7 +147,6 @@ public static <T, R extends AutoCloseable> Promise<T> tryApply(Promise<R> resour
                         }, true)
                        .unwrap();
     }
-    
 
     public static <T, R extends AsyncCloseable> Promise<T> tryApplyEx(CompletionStage<R> resourcePromise,
                                                                       Function<? super R, ? extends T> fn) {
@@ -221,18 +219,13 @@ public static <T, R extends AutoCloseable> Promise<T> tryCompose(Promise<R> reso
                                    return;
                                }
                                // CLOSE OK
-                               if (null == actionException) {
-                                   result.success(actionResult);
-                               } else {
-                                   result.failure(actionException);
-                               }
+                               result.complete(actionResult, actionException);
                            });
                            return result.onCancel(() -> cancelPromise(action, true));
                        }, true)
                        .unwrap();        
     }
 
-    
     public static <T, R extends AsyncCloseable> Promise<T> tryComposeEx(Promise<R> resourcePromise,
                                                                         Function<? super R, ? extends CompletionStage<T>> fn) {
         return
@@ -273,10 +266,8 @@ public static <T, R extends AsyncCloseable> Promise<T> tryComposeEx(Promise<R> r
                                            actionException.addSuppressed(onClose);
                                        }
                                        result.failure(actionException);
-                                   } else if (null != onClose) {
-                                       result.failure(onClose);
                                    } else {
-                                       result.success(actionResult);
+                                       result.complete(actionResult, onClose);
                                    }
                                });
                            });
@@ -315,7 +306,6 @@ public static <T, A, R> Promise<R> partitioned(Stream<? extends T> values,
         return partitioned2(values.iterator(), values, batchSize, spawner, downstream, downstreamExecutor);
     }
 
-    
     private static <T, A, R> Promise<R> partitioned1(Iterator<? extends T> values, 
                                                      Object source,
                                                      int batchSize, 
@@ -328,7 +318,6 @@ private static <T, A, R> Promise<R> partitioned1(Iterator<? extends T> values,
             .as(onCloseSource(null != source? source : values))
             .unwrap();
     }
-    
 
     private static <T, A, R> Promise<R> partitioned2(Iterator<? extends T> values, 
                                                      Object source,
@@ -400,7 +389,6 @@ private static <T, A, R> Promise<IndexedStep<A>> parallelStep2(
         });
     }
 
-    
     private static class IndexedStep<T> {
         private final int idx;
         private final T payload;
@@ -466,7 +454,6 @@ private static <T> Function<Promise<T>, Promise<T>> onCloseSource(Object source)
             return Function.identity();
         }
     }    
-
 
     /**
      * <p>Returns a promise that is resolved successfully when all {@link CompletionStage}-s passed as parameters
@@ -488,7 +475,6 @@ public static <T> Promise<List<T>> all(CompletionStage<? extends T>... promises)
         return all(Arrays.asList(promises));
     }
 
-    
     public static <T> Promise<List<T>> all(List<? extends CompletionStage<? extends T>> promises) {
         return all(true, promises);        
     }