building_data_pipelines_1_2.clj

(def name :capitalize-subscriber-name)

(def description 
  "Capitalize the subscriber name to increase the wow-factor.")

(def input (a/chan))

(def output (a/chan))

(def handler capitalize-subscriber-name)

(def process
  {:name name
   :description description
   :input input
   :output output
   :handler handler})

(defn call-fn-safely
  [f x]
  (try (f x)
       (catch java.lang.Throwable t t)))

(defn start-process
  [{:keys [name description input output handler] :as process}]
  (println (format "Starting Process: %s" name))
  (println (format "Description: %s" description))
  (a/go-loop [in-msg (a/<! input)]
    (when in-msg
      (when-let [output-data (call-fn-safely handler in-msg)]
        (if-not (instance? java.lang.Throwable output-data)
          (do (a/>! output {:data output-data})
              (recur (a/<! input)))
          (do (println "Stopping due to an unknown exception.")
              (println output-data)))))))
  
(start-process process)
  
(a/put! input {:data {:name "Sally"}})
  
; Should put output data to output chan
  
(a/poll! output)
  
; =>
  
; {:data {:name "SALLY"}}