Packages

gust

0.1.13
0.1.37 0.1.36 0.1.35 0.1.34 0.1.33 0.1.32 0.1.31 0.1.30 0.1.29 0.1.28 0.1.27 0.1.26 0.1.25 0.1.24 0.1.23 0.1.22 0.1.21 0.1.20 0.1.19 0.1.18 0.1.17 0.1.16 0.1.15 0.1.14 0.1.13 0.1.12 0.1.11 0.1.10 0.1.9 0.1.8 0.1.7 0.1.6 0.1.5 0.1.4 0.1.2 0.1.1 0.1.0

A DAG-Based Workflow Orchestration Engine for Elixir

HexDocs

Current section

Files

Jump to
Readme
37 Versions
9 Dependencies
Dependants
Files
Activity
gust lib gust dag scheduler worker.ex
Raw

lib/gust/dag/scheduler/worker.ex 

defmodule Gust.DAG.Scheduler.Worker do

  @behaviour Gust.DAG.Scheduler



  import Crontab.CronExpression

  alias Gust.DAG.Cron

  use GenServer

  alias Quantum.Job, as: QJob



  def init(args) do

    {:ok, args}

  end



  def start_link(args) do

    GenServer.start_link(__MODULE__, args, name: __MODULE__)

  end



  def schedule(dag_defs) do

    GenServer.call(__MODULE__, {:load_dags, dag_defs})

  end



  def handle_call({:load_dags, dag_defs}, _from, state) do

    jobs =

      for {dag_id, {:ok, dag_def}} <- dag_defs,

          schedule = dag_def.options[:schedule],

          schedule != nil,

          map_size(dag_def.error) == 0 do

        add_dag_job(dag_def, dag_id)

      end



    {:reply, jobs, state}

  end



  def add_dag_job(dag_def, dag_id) do

    schedule = dag_def.options[:schedule]



    Cron.new_job()

    |> QJob.set_name(String.to_atom(dag_def.name))

    |> QJob.set_schedule(~e[#{schedule}])

    |> QJob.set_task({Gust.DAG.RunRestarter, :start_dag, [dag_id]})

    |> Cron.add_job()

  end

end