      (ex-info message info cause))))
 
 
-(defn retry-error [s e]
+
+(defn retry-action [s e]
   (-> s
       (update-in [:retry :state] (fn [os] (if (not= :retry (:state s)) (:state s) os)))
       (update-in [:retry :attempts] (fn [ov] (inc (or ov 1))))

(defmethod transition :extract
  [s]
  (with-state-machine s
    :on-error   (dispatch-error
                 :errors/kafka-unavailable    retry-error
                 :errors/topic-not-found      retry-error
                 :errors/partition-not-found  retry-error
                 :errors/network-error        retry-error

                 ;; TODO: err: 4,5
                 ;;:errors/message-too-big      send-to-errors
                 ;;:errors/offset-not-available reset-offset
                 )
    :on-success (fn [s]
                  (if (empty? (get-in s [:data :batch]))
                    s
                    (clean-move-to :transform s)))

    (fn [s]
      (-> s
          extract-init-consumer

(ns samsara.qanal.state-machine
  (:require [safely.core :refer [sleeper]]))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;                                                                            ;;
;;        ---==| S T A T E   M A C H I N E   U T I L I T I E S |==----        ;;
;;                                                                            ;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

;;
;; This is the implementation of the state machine described
;; at doc/state-machine.md
;;

;; sample state
{
 ;;
 ;; the configuration of what this state machine
 ;; has to consume and where all the data
 ;; has to be loaded it is part of the `:config` element
 ;;
 :config {;; this is the connect string for the kafka
          ;; brokers, a comma-separated list
          ;; of host and ports
          :brokers ""

          ;; this is the topic and partition to consume
          :topic "topic1" :partition 2

          ;; the elasticsearch endpoint where
          ;; to load the data
          :elasticsearch ""
          }

 ;; TODO: signatures and doc
 :fns {:init-consumer (fn [cfg] :new-consumer)
       :check (fn [c cfg] {["topic1" 2] 2345})
       :poll (fn [c offset] [1 2 3 4])
       :parse    1
       :validate 1
       :tranform 1
       :init-load 1
       :bulk-load 1}

 :state :retry

