Data Flow Pipeline in Clojure

First, we can use exceptions. If it doesn't look like Clojure, we can some other constructs like below. Here, the building blocks are reduce and reduction. And specifically, it is the reduction function that helps to break the recursion if error occurs. You can also use monad based abstractions, which I don't prefer.

;; pipeline.clj

#!/usr/bin/env boot

(defn deps [new-deps]
  (merge-env! :dependencies new-deps))

(deps '[[org.clojure/clojure "1.8.0"]])

(defn validate-true [x]
  (println "validate-true")
  {:status true :state {:id 123 :foo "bar"}})

(defn validate-false [x]
  (println "validate-false")
  {:status false :code 1 :msg "error-validate-false"})

(defn validate-false-1 [x]
  (println "validate-false")
  {:status false :code 1 :msg "error-validate-false-1"})

(defn pipeline [fns]
  (reduce (fn [x fun]
    (if (:status x)
      (fun (:state x))
      (reduced x)))
    {:status true}
    fns))

(println "pipeline 0")
(println (pipeline [validate-false validate-true validate-false validate-true validate-true]))

(println "\npipeline 1")
(println (pipeline [validate-true validate-false-1 validate-true validate-false validate-true]))

(println "\npipeline 2")
(println (pipeline [validate-true validate-true validate-false validate-true validate-true]))

(println "\npipeline")
(println (pipeline [validate-true validate-true validate-true]))

;; output
pipeline 0
validate-false
{:status false, :code 1, :msg error-validate-false}

pipeline 1
validate-true
validate-false
{:status false, :code 1, :msg error-validate-false-1}

pipeline 2
validate-true
validate-true
validate-false
{:status false, :code 1, :msg error-validate-false}

pipeline
validate-true
validate-true
validate-true
{:status true}


# boot.properties
#https://github.com/boot-clj/boot
BOOT_CLOJURE_NAME=org.clojure/clojure
BOOT_VERSION=2.7.1
BOOT_CLOJURE_VERSION=1.8.0

The pipeline accepts a list of functions and chains them together. Each reduction can pass in the required arguments necessary for the next function in the chain. These functions must be composable to make this work. The reduced function checks for a recursion termination condition apart from the reduction termination. When the condition which is the {:status false ..} is met, it short circuits the evaluation and exits the recursion, returning the value of the computation. This value can then be matched and responded accordingly. In case of web app, a suitable ring response can be generated.

The pipeline can be modified to accept a collection of [ok-fn err-fn] so that instead of having a generic one error handler, we can have two different paths and localised handling of errors.

(defn pipeline [fns]
  (reduce (fn [x fun]
            (if (:status x)
              (if (= (type fun) clojure.lang.PersistentVector)
                (((first fun) (:state x)))
                (fun (:state x)))
              (if (= (type fun) clojure.lang.PersistentVector)
                (reduced ((second fun) x))
                (reduced x))))
          {:status true}
          fns))

(defn validate-false-s1 [x]
  (println "special error handling fun in pipeline")
  {:status false :code -1 :msg "nop"}) 

(println "\npipeline separate flows")
(println (pipeline [validate-true validate-false [validate-true validate-false-s1] validate-true]))

; output
pipeline separate flows
validate-true
validate-false
special error handling fun in pipeline
{:status false, :code -1, :msg nop}

The pipeline is very flexible and can be modified to suite variety of use cases. The contract does not always have to be a map with :status. We can use core.match to do various types of conditional branching.

PS: Using functional language does not mean that whole code will be functional. There will be a few parts that are less functional, and that is totally fine. The point here is to be pragmatic. The main use of monad is to sequence code.