Packages

gust

0.1.17
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 runner task_worker.ex
Raw

lib/gust/dag/runner/task_worker.ex 

defmodule Gust.DAG.Runner.TaskWorker do

  @moduledoc false

  use GenServer

  alias Gust.DAG



  @impl true

  def init(init_arg) do

    {:ok, init_arg, {:continue, :init_run}}

  end



  def start_link(args) do

    GenServer.start_link(__MODULE__, args, name: via_tuple("task_#{args[:task].id}"))

  end



  defp via_tuple(name) do

    {:via, Registry, {Gust.Registry, name}}

  end



  def child_spec(args) do

    %{

      id: __MODULE__,

      start: {__MODULE__, :start_link, [args]},

      restart: :temporary,

      type: :worker

    }

  end



  @impl true

  def handle_continue(:init_run, state) do

    send(self(), :run)



    {:noreply, state}

  end



  @impl true

  def handle_info(

        :run,

        %{task: task, mod: mod, stage_pid: stage_pid} = state

      ) do

    fun_name = String.to_atom(task.name)

    args = [%{run_id: task.run_id}]



    DAG.Logger.set_task(task.id, task.attempt)



    {status, result} =

      case try_run(mod, fun_name, args) do

        {:ok, result} ->

          {:ok, result}



        {:error, error} ->

          {:error, error}

      end



    DAG.Logger.unset()



    send(stage_pid, {:task_result, result, task.id, status})



    {:stop, :normal, state}

  end



  defp try_run(mod, fun_name, args) do

    {:ok, apply(mod, fun_name, args)}

  rescue

    e -> {:error, e}

  end

end