Packages
step_flow
1.6.0-rc15
1.9.0-rc2
1.9.0-rc1
1.9.0-rc0
1.8.2
1.8.1
1.8.1-rc8
1.8.1-rc7
1.8.1-rc6
1.8.1-rc5
1.8.1-rc4
1.8.1-rc3
1.8.1-rc2
1.8.1-rc1
1.8.1-rc0
1.8.0
1.8.0-rc3
1.8.0-rc2
1.8.0-rc1
1.8.0-rc0
1.7.3
1.7.3-rc4
1.7.3-rc3
1.7.3-rc2
1.7.3-rc1
1.7.3-rc0
1.7.2
1.7.2-rc4
1.7.2-rc3
1.7.2-rc2
1.7.2-rc1
1.7.2-rc0
1.7.1
1.7.0
1.7.0-rc1
1.7.0-rc0
1.6.1
1.6.1-rc1
1.6.1-rc0
1.6.0
1.6.0-rc9
1.6.0-rc8
1.6.0-rc7
1.6.0-rc6
1.6.0-rc5
1.6.0-rc4
1.6.0-rc3
1.6.0-rc20
1.6.0-rc2
1.6.0-rc19
1.6.0-rc18
1.6.0-rc17
1.6.0-rc16
1.6.0-rc15
1.6.0-rc14
1.6.0-rc13
1.6.0-rc12
1.6.0-rc11
1.6.0-rc10
1.6.0-rc1
1.5.0
1.5.0-rc1
1.4.2-rc2
1.4.2-rc1
1.4.1
1.4.1-rc1
1.4.0
1.4.0-rc4
1.4.0-rc3
1.4.0-rc2
1.4.0-rc1
1.3.1
1.3.0
1.3.0-rc
1.2.0
1.1.0
1.0.0
1.0.0-rc9
1.0.0-rc8
1.0.0-rc7
1.0.0-rc6
1.0.0-rc5
1.0.0-rc1
0.2.13
0.2.12
0.2.11
0.2.10
0.2.9
0.2.8
0.2.7
0.2.6
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.24
0.0.23
0.0.22
0.0.21
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14
0.0.13
0.0.12
0.0.11
0.0.10
0.0.9
0.0.8
0.0.7
0.0.6
0.0.4
0.0.3
0.0.2
0.0.1
Step flow manager for Elixir applications
Current section
Files
Jump to
Current section
Files
lib/step_flow/amqp/worker_terminated_consumer.ex
defmodule StepFlow.Amqp.WorkerTerminatedConsumer do
@moduledoc """
Consumer of all worker terminations.
"""
require Logger
alias StepFlow.Amqp.WorkerTerminatedConsumer
alias StepFlow.LiveWorkers
alias StepFlow.Repo.Checker
alias StepFlow.Workers.WorkerStatusWatcher
alias StepFlow.Workflows
alias StepFlow.Workflows.StepManager
use StepFlow.Amqp.CommonConsumer, %{
queue: "worker_terminated",
exchange: "worker_response",
prefetch_count: 1,
consumer: &WorkerTerminatedConsumer.consume/4
}
@doc """
Consume worker terminated message.
"""
def consume(
channel,
tag,
_redelivered,
%{
"job_id" => job_id
} = _payload
) do
live_worker_update(job_id)
Basic.ack(channel, tag)
end
def consume(
channel,
tag,
_redelivered,
%{
"instance_id" => instance_id
} = payload
) do
if Checker.repo_running?() do
Logger.info("Worker #{instance_id} terminated #{inspect(payload)}")
WorkerStatusWatcher.update_worker_status(instance_id, %{
instance_id: instance_id,
activity: "Terminated"
})
Basic.ack(channel, tag)
else
Logger.warn(
"#{__MODULE__}: The database is not available, reject and requeue consumed message..."
)
Basic.reject(channel, tag, requeue: true)
end
end
def consume(channel, tag, _redelivered, payload) do
Logger.error("Worker terminated #{inspect(payload)}")
Basic.reject(channel, tag, requeue: false)
end
defp live_worker_update(job_id) do
live_worker = LiveWorkers.get_by(%{"job_id" => job_id})
LiveWorkers.update_live_worker(live_worker, %{
"termination_date" => NaiveDateTime.utc_now()
})
Workflows.notification_from_job(job_id)
StepManager.check_step_status(%{job_id: job_id})
:ok
end
end