ASYNC-275: Changed futurize return Future from CompletableFuture to FutureTask to match the same type returned from ExecutorService.submit. Added a regression test to catch this case.

Author: fogus

src/main/clojure/clojure/core/async/flow/impl.clj

@@ -14,7 +14,7 @@
             [clojure.core.async.impl.dispatch :as disp]
             [clojure.walk :as walk]
             [clojure.datafy :as datafy])
-  (:import [java.util.concurrent Future Executor TimeUnit CompletableFuture]
+  (:import [java.util.concurrent Future Executor TimeUnit FutureTask]
            [java.util.concurrent.locks ReentrantLock]))
 
 (set! *warn-on-reflection* true)
@@ -31,8 +31,8 @@
     (let [^Executor e (if (instance? Executor exec)
                         exec
                         (disp/executor-for exec))
-          fut (CompletableFuture.)]
-      (.execute e #(.complete fut (apply f args)))
+          fut (FutureTask. #(apply f args))]
+      (.execute e fut)
       fut)))
 
 (defn prep-proc [ret pid {:keys [proc, args, chan-opts] :or {chan-opts {}}}]

src/test/clojure/clojure/core/async/flow_test.clj

@@ -5,20 +5,28 @@
 ;;   By using this software in any fashion, you are agreeing to be bound by
 ;;   the terms of this license.
 ;;   You must not remove this notice, or any other, from this software.
-
 (ns clojure.core.async.flow-test
   (:require [clojure.test :refer :all]
-            [clojure.core.async.flow :as flow]))
+            [clojure.core.async.flow :as flow])
+  (:import [java.util.concurrent TimeUnit Future ExecutionException]))
+
+(set! *warn-on-reflection* true)
 
 (deftest test-futurize
-  (testing ""
+  (testing "Happy path use of futurize"
     (let [in-es? (atom false)
           es (reify java.util.concurrent.Executor
-               (^void execute [_ ^Runnable f]
-                 (reset! in-es? true)
-                 (future-call f)))]
+               (^void execute [_ ^Runnable r]
+                (reset! in-es? true)
+                (future (.run r))))]
       (is (= 16 @((flow/futurize #(* % %) {:exec :mixed}) 4)))
       (is (= 16 @((flow/futurize #(* % %)) 4)))
       (is (= 16 @((flow/futurize #(* % %) {:exec es}) 4)))
-      (is @in-es?))))
-
+      (is @in-es?)))
+  (testing "ASYNC-275 regression: futurize Future instance propagates exceptions"
+    (let [cause (ex-info "boom" {})
+          fut   ((flow/futurize (fn [] (throw cause)) {:exec :mixed}))
+          ex    (try
+                  (.get ^Future fut 1000 TimeUnit/MILLISECONDS)
+                  (catch ExecutionException e e))]
+      (is (= cause (.getCause ^ExecutionException ex))))))