fix :workload :compute

Author: richhickey

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

@@ -252,8 +252,7 @@
 
   When :compute is specified transform must not block!"
   ([fn-or-map] (process fn-or-map nil))
-  ([fn-or-map {:keys [workload timeout-ms]
-               :or {timeout-ms 5000} :as opts}]
+  ([fn-or-map {:keys [workload compute-timeout-ms] :as opts}]
    (impl/proc fn-or-map opts)))
 
 (defn lift*->step

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

@@ -26,7 +26,7 @@
     clojure.lang.Var (symbol x)
     (datafy/datafy x)))
 
-(defn futurize ^Future [f {:keys [exec]}]
+(defn futurize [f {:keys [exec]}]
   (fn [& args]
     (let [^ExecutorService e (if (instance? ExecutorService exec)
                                exec
@@ -219,7 +219,7 @@
 
 (defn proc
   "see lib ns for docs"
-  [fm {:keys [workload compute-timeout-ms]}]
+  [fm {:keys [workload compute-timeout-ms] :or {compute-timeout-ms 5000}}]
   (let [{:keys [describe init transition transform] :as impl}
         (if (map? fm) fm {:describe fm :init fm :transition fm :transform fm})
         {:keys [params ins] :as desc} (describe)
@@ -236,10 +236,10 @@
       (describe [_] desc)
       (start [_ {:keys [pid args ins outs resolver]}]
         (assert (or (not params) args) "must provide :args if :params")
-        (let [comp? (= workload :compute)
-              transform (cond-> transform (= workload :compute)
-                                #(.get (futurize transform {:exec (spi/get-exec resolver :compute)})
-                                       compute-timeout-ms TimeUnit/MILLISECONDS))
+        (let [transform (if (= workload :compute)
+                          #(.get ^Future ((futurize transform {:exec (spi/get-exec resolver :compute)}) %1 %2 %3)
+                                 compute-timeout-ms TimeUnit/MILLISECONDS)
+                          transform)
               exs (spi/get-exec resolver (if (= workload :mixed) :mixed :io))
               state (when init (init args))
               ins (into (or ins {}) (::flow/in-ports state))