Add ThenCompose CompletionStage method

Author: cosineblast

src/manifold/deferred.clj

@@ -64,7 +64,9 @@
 
          apply-to-either apply-to-either-async
          accept-either accept-either-async
-         run-after-either run-after-either-async)
+         run-after-either run-after-either-async
+
+         then-compose then-compose-async)
 
 ;; The potemkin abstract type for
 ;; implementations such as CompletionStage
@@ -132,7 +134,14 @@
   (runAfterEitherAsync [this operator other]
     (run-after-either-async this operator other))
   (runAfterEitherAsync [this operator other executor]
-    (run-after-either-async this operator other executor)))
+    (run-after-either-async this operator other executor))
+
+  (thenCompose [this operator]
+    (then-compose this operator))
+  (thenComposeAsync [this operator]
+    (then-compose-async this operator))
+  (thenComposeAsync [this operator executor]
+    (then-compose-async this operator executor)))
 
 (definline realized?
   "Returns true if the manifold deferred is realized."
@@ -1575,6 +1584,12 @@
 
 (def ^:private run-after-either-async (async-for-dual run-after-either))
 
+(defn- then-compose [this ^java.util.function.Function operator]
+  (assert-some operator)
+  (chain this (fn [value] (.apply operator value))))
+
+(def ^:private then-compose-async (async-for then-compose))
+
 
 
 

test/manifold/deferred_stage_test.clj

@@ -3,6 +3,7 @@
             [clojure.test :refer [deftest is testing]])
   (:import [java.util.concurrent
             CompletionStage
+            CompletableFuture
             Executors]))
 
 (defn fn->Function [function]
@@ -323,3 +324,66 @@
       (dorun (for [method-info alt-method-info
                    mode [:raw :async :with-executor]]
                (test-alt-error method-info mode executor))))))
+
+
+
+(def compose-method-info
+  {:methods {:raw
+              (fn [^CompletionStage this operator _]
+                (.thenCompose this operator))
+              :async
+              (fn [^CompletionStage this operator _]
+                (.thenComposeAsync this operator))
+              :with-executor
+              (fn [^CompletionStage this operator executor]
+                (.thenComposeAsync this operator executor))}
+    :interface fn->Function
+    :inner-assertion #(is (= 1 %))
+    :post-assertion #(is (= 2 %))})
+
+(defn- test-compose-success [method-info mode executor]
+
+  (let [was-called (atom false)
+
+        method (get-in method-info [:methods mode])
+        {:keys [inner-assertion post-assertion]
+         to-java-interface :interface} method-info
+
+        d1 (d/success-deferred 1)
+        d2 (method
+            d1
+            (to-java-interface
+             (fn [x]
+               (inner-assertion x)
+               (reset! was-called true)
+               (d/success-deferred 2)))
+            executor)]
+
+    (is (= @d1 1))
+    (post-assertion @d2)
+    (is (= true @was-called))))
+
+
+(deftest test-compose
+
+  (let [executor (Executors/newSingleThreadExecutor)]
+    (testing "compose success"
+      (dorun (for [method-info [compose-method-info]
+                   mode [:raw :async :with-executor]]
+               (test-compose-success method-info mode executor))))
+
+    (testing "compose error"
+      (dorun (for [method-info [compose-method-info]
+                   mode [:raw :async :with-executor]]
+               (test-functor-error method-info mode executor))))))
+
+(deftest test-compose-into-completable-future
+
+  (testing "deferred can compose into CompletableFuture"
+    (let [d1 ^CompletionStage (d/success-deferred 10)
+          d2 (.thenCompose
+              d1
+              (fn->Function
+               (fn [x] (CompletableFuture/completedFuture (inc x)))
+               ))]
+      (is (= @d2 11)))))