ASYNC-270: Modified the operation of futurize to account for the change to Executor interface in ASYNC-272. The futurize function now creates and returns future. Also, merging in options so that exeception is not thrown when :exec not supplied.

Author: fogus

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

@@ -30,8 +30,10 @@
   (fn [& args]
     (let [^Executor e (if (instance? Executor exec)
                         exec
-                        (disp/executor-for exec))]
-      (.execute e ^Runnable #(apply f args)))))
+                        (disp/executor-for exec))
+          fut (java.util.concurrent.CompletableFuture.)]
+      (.execute e ^Runnable #(.complete fut (apply f args)))
+      fut)))
 
 (defn prep-proc [ret pid {:keys [proc, args, chan-opts] :or {chan-opts {}}}]
   (let [{:keys [ins outs signal-select]} (spi/describe proc)

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

@@ -12,10 +12,13 @@
 
 (deftest test-futurize
   (testing ""
-    (let [es (reify java.util.concurrent.ExecutorService
-               (^java.util.concurrent.Future submit [_ ^Callable f]
-                (future-call (comp vector f))))]
+    (let [in-es? (atom false)
+          es (reify java.util.concurrent.Executor
+               (^void execute [_ ^Runnable f]
+                 (reset! in-es? true)
+                 (future-call f)))]
       (is (= 16 @((flow/futurize #(* % %) {:exec :mixed}) 4)))
       (is (= 16 @((flow/futurize #(* % %)) 4)))
-      (is (= [16] @((flow/futurize #(* % %) {:exec es}) 4))))))
+      (is (= 16 @((flow/futurize #(* % %) {:exec es}) 4)))
+      (is @in-es?))))