   (let [a (streams/stream 10)
         b (streams/stream 10)
         c (streams/stream 10)
-        splitter (output-splitter a :multi-thread)
+        tp (threadpool 4)
+        splitter (output-splitter a :single-thread)
         read (fn [x y]
                (streams/consume
                 (fn [v]
-                  (Thread/sleep 100)
-                  (println (format "%s-%s: %s" (.getName (Thread/currentThread)) y v))) x))]
+                  (ha/future-with tp
+                                  (Thread/sleep 100)
+                                  (println (format "%s-%s: %s" (.getName (Thread/currentThread)) y v)))) x))]
 
     (proto/connect splitter b op/default-operations)
     (proto/connect splitter c op/default-operations)

                           (streams/put! sink tv)))
                       sink))

(defn- connect-threaded
  [source sink op-map]
  (streams/connect-via source
                       (fn [v]
                         (ha/future
                           (when-let [tv (transform-and-put v op-map)]
                             (streams/put! sink tv))))
                       sink))


(deftype OutputSplitter [source sinks]
  proto/Splitter

  ([source execution-mode]
   (cond
     (= execution-mode :single-thread) (OutputSplitter. source (atom []))
     (= execution-mode :multi-thread) (ThreadedOutputSplitter. source (atom []) (threadpool 4))
     :else (throw (ex-info (format "Don't know about '%s' execution mode."
                                   execution-mode)
                           :cause {:execution-mode execution-mode})))))
