Packages
step_flow
1.6.0-rc17
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/workflows/workflows_watcher.ex
defmodule StepFlow.Workflows.WorkflowsWatcher do
require Logger
@moduledoc false
use GenServer
alias StepFlow.Workflows
alias StepFlow.Workflows.Status
# 10 seconds in milliseconds
@default_workflows_check_interval 10_000
def child_spec(_) do
%{
id: StepFlow.Workflows.WorkflowsWatcher,
start: {StepFlow.Workflows.WorkflowsWatcher, :start_link, []},
type: :worker
}
end
def start_link do
Logger.warn("#{__MODULE__} start_link")
GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
end
@impl true
def init(params \\ %{}) do
Logger.warn("[#{__MODULE__}] Start checking workflows!")
check_workflows()
{:ok, params}
end
@impl true
def handle_info(:check, content) do
Logger.info("[#{__MODULE__}] Check workflows status...")
# Schedule once more
check_workflows()
{:noreply, content}
end
@impl true
def handle_call(:pop, _id, _content) do
{:reply, nil}
end
@impl true
def handle_cast({:push, _id, content}, _params) do
check_workflows()
{:noreply, content}
end
defp check_workflows do
# Wait for the AMQP supervisor to be started before checking workflows
if Process.whereis(StepFlow.Amqp.Supervisor) != nil do
check_paused_workflows()
end
interval =
Application.get_env(:step_flow, Workflows,
workflows_check_interval: @default_workflows_check_interval
)
|> Keyword.get(:workflows_check_interval, @default_workflows_check_interval)
Process.send_after(self(), :check, interval)
end
defp check_paused_workflows do
now = DateTime.from_unix!(System.os_time(:second))
get_paused_workflows()
|> Enum.each(fn workflow ->
trigger_post_action_if_status_expired(workflow, now)
end)
end
defp get_paused_workflows do
params = %{
"states" => ["paused", "pausing"],
"size" => 1000,
"roles" => ["administrator"]
}
Workflows.list_workflows(params)
|> Map.get(:data)
|> Enum.filter(fn workflow ->
last_status = Status.get_last_workflow_status(workflow.id)
last_status.state in [:paused, :pausing]
end)
end
defp trigger_post_action_if_status_expired(workflow, now) do
case Status.get_last_workflow_status(workflow.id)
|> Map.get(:description) do
nil ->
nil
description ->
Logger.debug(
"Workflow #{workflow.id} is paused. Last status description: #{inspect(description)}"
)
should_trigger =
StepFlow.Map.get_by_key_or_atom(description, :trigger_at)
|> get_trigger_date!()
|> should_trigger?(now)
if should_trigger do
StepFlow.Map.get_by_key_or_atom(description, :action)
|> trigger_post_action(workflow)
end
end
end
defp get_trigger_date!(trigger_at) when is_integer(trigger_at) do
DateTime.from_unix!(trigger_at, :millisecond)
|> DateTime.to_naive()
end
defp get_trigger_date!(trigger_at) when is_bitstring(trigger_at) do
NaiveDateTime.from_iso8601!(trigger_at)
end
defp get_trigger_date!(trigger_at), do: trigger_at
defp should_trigger?(nil, _now), do: false
defp should_trigger?(trigger_date, now) do
NaiveDateTime.diff(trigger_date, now) < 0
end
defp trigger_post_action(post_action, workflow) do
case post_action do
"resume" ->
Logger.info("Resume paused workflow #{workflow.id}...")
case Workflows.resume(workflow) do
{:ok, _} ->
{}
{:error, message} ->
Logger.error(
"Something went wrong resuming workflow #{workflow.id} #{workflow.identifier}: #{inspect(message)}"
)
end
"abort" ->
Logger.warn("Abort paused workflow #{workflow.id}...")
case Workflows.abort(workflow) do
{:ok, _} ->
{}
{:error, message} ->
Logger.error(
"Something went wrong aborting workflow #{workflow.id} #{workflow.identifier}: #{inspect(message)}"
)
end
_ ->
Logger.warn("Invalid paused status action: #{inspect(post_action)}")
{}
end
end
end