ASYNC-263: Throwing when :chan-opts on out chan.

Author: fogus

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

@@ -73,7 +73,9 @@
   :args - a map of param->val which will be passed to the process ctor
   :chan-opts - a map of in-or-out-id->{:keys [buf-or-n xform]}, where buf-or-n
                and xform have their meanings per core.async/chan
-               the default is {:buf-or-n 10}
+               the default is {:buf-or-n 10}. For efficiency, Flow creates
+               only the in channel per in/out pair and so :chan-opts
+               should be associated with the in channel.
   
   :conns - a collection of [[from-pid outid] [to-pid inid]] tuples.
 

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

@@ -62,7 +62,11 @@
         conn-map (reduce (fn [ret [out in :as conn]]
                            (if (and (contains? outopts out)
                                     (contains? inopts in))
-                             (update ret out set-conj in)
+                             (if (seq (vals (get outopts out)))
+                               (throw (ex-info (str "only one channel created for connection. "
+                                                    ":chan-opts must be associated with input side.")
+                                               {:conn conn, :out-opts outopts}))
+                               (update ret out set-conj in))
                              (throw (ex-info "invalid connection" {:conn conn}))))
                          {} conns)
         running-chans #(or (deref chans) (throw (Exception. "flow not running")))

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

@@ -0,0 +1,21 @@
+(ns clojure.core.flow-test
+  (:require [clojure.core.async.flow :as flow]
+            [clojure.test :refer :all]))
+
+(defn tap-n-drop [x] (tap> x) nil)
+
+(deftest chan-opts-tests
+  (testing ":chan-opts only specified on in side of connected pair"
+    (is (thrown? clojure.lang.ExceptionInfo
+                 (flow/create-flow
+                  {:procs {:source {:proc (-> identity flow/lift1->step flow/process)
+                                    :chan-opts {:out {:buf-or-n 11
+                                                      :xform (map (fn [x] (str "Saw " x)))}}}
+                           :sink {:proc (-> #'tap-n-drop flow/lift1->step flow/process)}}
+                   :conns [[[:source :out] [:sink :in]]]})))
+    (is (flow/create-flow
+         {:procs {:source {:proc (-> identity flow/lift1->step flow/process)}
+                  :sink {:proc (-> #'tap-n-drop flow/lift1->step flow/process)
+                         :chan-opts {:in {:buf-or-n 11
+                                          :xform (map (fn [x] (str "Saw " x)))}}}}
+          :conns [[[:source :out] [:sink :in]]]}))))