Merge pull request #3 from ElChache/should-anomaly-break

is-anomaly-error

project.clj

@@ -1,4 +1,4 @@
-(defproject elchache/fonda "0.0.1"
+(defproject elchache/fonda "0.0.2-SNAPSHOT"
   :url "https://github.com/arichiardi/fonda"
   :description "An async pipeline approach to functional core - imperative shell."
   :license {:name "Apache License"

src/fonda/execute.cljs

@@ -16,11 +16,13 @@
             anomaly? (update :processor-results-stack conj res))))
 
 (defn assoc-processor-result
-  [{:as fonda-ctx :keys [anomaly-fn]} path res]
+  [{:as fonda-ctx :keys [anomaly-fn]}
+   {:keys [path is-anomaly-error?]}
+   res]
   (let [new-fonda-ctx (cond
 
                         ;; If it is an anomaly, associates the anomaly on the context, and execution will stop here
-                        (and anomaly-fn (anomaly-fn res)) (assoc fonda-ctx :anomaly res)
+                        (and anomaly-fn (anomaly-fn res) (is-anomaly-error? res)) (assoc fonda-ctx :anomaly res)
 
                         ;; If there is no path on the step, it doesn't contribute to the context
                         (nil? path) fonda-ctx
@@ -45,18 +47,21 @@
 (defn handle-exception
   [{:as fonda-ctx :keys [ctx]} {:keys [on-error] :as step} e]
   (when on-error (on-error e ctx))
+  (when (:name step) (println "Exception on step " (:name step)))
   (assoc-exception-result fonda-ctx e))
 
 (defn invoke-post-callback-fns
-  [{:as fonda-ctx :keys [anomaly-fn ctx]} {:keys [on-complete on-success on-error path] :as step} step-res]
+  [{:as fonda-ctx :keys [anomaly-fn ctx]}
+   {:keys [on-complete on-success on-error path is-anomaly-error?] :as step}
+   step-res]
 
   (let [aug-ctx (if path (assoc-in ctx path step-res) ctx)]
 
     ;; Always calls on-complete
     (when on-complete
       (on-complete step-res aug-ctx))
 
-    (if (and anomaly-fn (anomaly-fn step-res))
+    (if (and anomaly-fn (anomaly-fn step-res) #_(is-anomaly-error? step-res))
 
       ;; If anomaly, calls on-error
       (when on-error (on-error step-res aug-ctx))
@@ -77,7 +82,7 @@
     (let [last-res (last processor-results-stack)
 
           ;; First step only gets the ctx, next ones receive last-result,ctx
-          args (if (empty? processor-results-stack) [ctx] [last-res ctx])
+          args (if (empty? processor-results-stack) [ctx] [ctx last-res])
 
           ; fn is an alias for processor
           processor (or processor (:fn step))
@@ -88,7 +93,7 @@
           res (apply f args)
           assoc-result-fn (cond
                             tap (partial assoc-tap-result fonda-ctx)
-                            processor (partial assoc-processor-result fonda-ctx (:path step))
+                            processor (partial assoc-processor-result fonda-ctx step)
                             inject (partial assoc-injector-result fonda-ctx))]
 
       ;; Invokes the callback functions
@@ -112,7 +117,7 @@
     (let [[cb result] (cond exception [(:on-exception fonda-ctx) exception]
                             anomaly [(:on-anomaly fonda-ctx) anomaly]
                             :else [(:on-success fonda-ctx) ctx])]
-      (cb (last processor-results-stack) result))))
+      (cb result (last processor-results-stack)))))
 
 (defn execute-steps
   "Sequentially runs each of the steps.

src/fonda/execute/specs.cljc

@@ -65,7 +65,7 @@
 
 (s/fdef fonda.execute/assoc-processor-result
   :args (s/cat :fonda-ctx ::fonda-context
-               :path ::step/path
+               :step ::step/processor-step
                :res any?))
 
 (s/fdef fonda.execute/assoc-tap-result

src/fonda/step.cljs

@@ -7,7 +7,16 @@
    tap
 
    ;; The name for the step
-   name])
+   name
+
+   on-start
+
+   on-success
+
+   on-error
+
+   on-complete
+   ])
 
 (defrecord Processor
   [;; A function that gets the context, the result is attached to the context on the given path
@@ -17,7 +26,18 @@
    name
 
    ;; Path were to attach the processor result on the context
-   path])
+   path
+
+   on-start
+
+   on-success
+
+   on-error
+
+   on-complete
+
+   is-anomaly-error?
+   ])
 
 (defrecord Injector
   [;; Function that returns step(s) to be injected  right after this step on the queue
@@ -37,7 +57,9 @@
   (let [step
         (cond
           tap (map->Tap (update step :tap resolve-function))
-          (:fn step) (map->Processor (update step :fn resolve-function))
+          (:fn step) (map->Processor (-> step
+                                         (update :fn resolve-function)
+                                         (assoc :is-anomaly-error? (or (:is-anomaly-error? step) (constantly true)))))
           inject (map->Injector (update step :inject resolve-function)))]
     (update step :name keyword)))
 

src/fonda/step/specs.cljc

@@ -1,16 +1,29 @@
 (ns fonda.step.specs
   (:require [clojure.spec.alpha :as s]))
 
+(s/def ::on-start (s/nilable fn?))
+(s/def ::on-success (s/nilable fn?))
+(s/def ::on-error (s/nilable fn?))
+(s/def ::on-complete (s/nilable fn?))
+(s/def ::is-anomaly-error? (s/nilable fn?))
+
 ;; Tap step
 (s/def ::tap (s/or :function fn? :qualified-keyword qualified-keyword?))
 (s/def ::tap-step
-  (s/keys :req-un [::tap]))
+  (s/keys :req-un [::tap]
+          :opt-un [::on-start ::on-success ::on-error ::on-complete]))
 
 ;; Processor step
-(s/def ::path vector?)
+(s/def ::path (s/nilable vector?))
 (s/def ::fn (s/or :function fn? :qualified-keyword qualified-keyword?))
 (s/def ::processor-step
-  (s/keys :req-un [::fn ::path]))
+  (s/keys :req-un [::fn]
+          :opt-un [::path
+                   ::on-start
+                   ::on-success
+                   ::on-error
+                   ::on-complete
+                   ::is-anomaly-error?]))
 
 ;; Injector step
 (s/def ::inject (s/or :function fn? :qualified-keyword qualified-keyword?))