titanoboa.processor — io.titanoboa/titanoboa 0.8.2

*cmd-exchange-ch* clj

*server-config* clj

add->history clj

(add->history job {:keys [step-end end message exception result] :as props})

assess4retry clj

(assess4retry {:keys [retry-on-error? max-retries id] :as step}
              step-retries-map)

contains-some clj

(contains-some pred coll)

Returns the first non-nil value of (pred x) for any x in coll, else nil. One common idiom is to use a set as pred, for example this will return :fred if :fred is in the sequence, otherwise nil: (some #{:fred} coll)

dispatch-job-threads! clj

(dispatch-job-threads!
  out-jobs-ch
  {:keys [step next-step parallel-threads thread-call-stack] :as job})

dispatch4join! clj

(dispatch4join! {:keys [thread-stack parallel-threads] :as job})

dispatch4join? clj

(dispatch4join? {:keys [thread-stack] :as job})

finalize-job! clj

(finalize-job! {:keys [jobid thread-stack callback-ch] :as job}
               finished-ch
               ack-fns-vec
               update-cache-fn
               &
               [commit-callback-ch])

find-matching-steps clj

(find-matching-steps steps r)

find-next-step clj

(find-next-step current-step step-result)

get-prop-trimming-fn clj

(get-prop-trimming-fn dont-log-properties
                      trim-logged-properties
                      properties-trim-size)

get-step clj

(get-step {{:keys [steps] :as jobdef} :jobdef :as job} step-id)

get-thread-id clj

(get-thread-id {:keys [thread-stack] :as job})

init-first-step clj

(init-first-step {{:keys [steps] :as jobdef} :jobdef
                  :keys [state next-step jobdir properties step-start]
                  :as job})

init-step clj

(init-step {{:keys [steps] :as jobdef} :jobdef
            :keys [state step-start next-step jobdir properties step-retries
                   node-id]
            :as job})

initialize-step clj

(initialize-step {:keys [state] :as job})

instantiate-job! clj

(instantiate-job! {:keys [tracking-id id jobdef jobdef-name revision properties
                          files new-jobs-ch state-agent job-folder defs-atom
                          mq-pool callback-ch]
                   :as config})

new-history-stub clj

(new-history-stub {:keys [step step-start] :as job})

next-step-join? clj

(next-step-join? job next-step-id)

normalize-result clj

(normalize-result step-result)

orchestrate-join! clj

(orchestrate-join! {:keys [thread-stack parallel-threads step] :as job})

orchestrate-join? clj

(orchestrate-join? {:keys [thread-stack] :as job})

orchestrate-step clj

(orchestrate-step {:keys [step thread-stack] :as job} node-id)

preprocess clj

(preprocess job)

process-map-step clj

(process-map-step {{:keys [jobdef-name sys-key standalone-system?
                           distributed-system? workers-count aggregator-q-name
                           create-folder?]
                    :as properties}
                     :properties
                   {step-id :id} :step
                   map-steps :map-steps
                   jobdir :jobdir
                   jobid :jobid
                   :as job})

process-reduce-step clj

(process-reduce-step {{:keys [terminate-standalone? map-step-id commit-interval]
                       :as properties}
                        :properties
                      {step-id :id :as step} :step
                      map-steps :map-steps
                      jobdir :jobdir
                      jobid :jobid
                      :as job})

process-step clj

(process-step {:keys [state step step-start start map-steps reduce-steps
                      thread-stack step-retries]
               :as job}
              node-id)

run-job! clj

(run-job! system-key
          {:keys [jobdef jobdef-name revision properties files] :as conf}
          sync)

Dispatches a request to start a job in given system. Does not use action thread pool. Returns the job id or the finished job if the sync flag is set to true.

run-sync-job! clj

(run-sync-job! {:keys [new-jobs-ch mq-pool] :as config})

start-job! clj

(start-job! system-key {:keys [jobdef jobdef-name properties files] :as conf})

(start-job! system-key
            {:keys [jobdef jobdef-name revision properties files] :as conf}
            response-ch
            keep-open)

Dispatches an action request to start a job in given system. The request is dispatched to the system's action thread pool. If a response channel is provided (it has to be a core async channel, distributed channels are not accepted!) the job's id will be put on the channel once the job has been instantiated. If the response channel is not provided, the function waits synchronously for the job to be instantiated and then returns the job id.

start-processor! clj

(start-processor!
  {:keys [stop-chan in-jobs-ch new-jobs-ch out-jobs-ch finished-ch state-agent
          eviction-agent mq-session node-id cmd-exchange-ch server-config
          dont-log-properties trim-logged-properties properties-trim-size]
   :as config})

trim-stack clj

(trim-stack job)