Packages

gust

0.1.31
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 task_worker adapters elixir.ex
Raw

lib/gust/dag/task_worker/adapters/elixir.ex 

defmodule Gust.DAG.TaskWorker.Adapters.Elixir do

  @moduledoc false



  use Gust.DAG.TaskWorker

  alias Gust.DAG.Logger



  @impl true

  def handle_info(

        :run,

        %{task: task, dag_def: dag_def, stage_pid: stage_pid, opts: opts} = state

      ) do

    fun_name = String.to_atom(task.name)

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



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



    {status, result} =

      case try_run(dag_def.mod, fun_name, args, opts[:store_result]) do

        {:ok, result} ->

          {:ok, result}



        {:error, error} ->

          {:error, error}

      end



    Logger.unset()



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



    {:stop, :normal, state}

  end



  defp try_run(mod, fun_name, args, store_result) do

    apply_and_validate(mod, fun_name, args, store_result)

  rescue

    e -> {:error, e}

  end



  defp apply_and_validate(mod, fun_name, args, store_result) do

    result = apply(mod, fun_name, args)

    maybe_validate_result(store_result, result)

  end



  def maybe_validate_result(false, result), do: {:ok, result}

  def maybe_validate_result(true, result) when is_map(result), do: {:ok, result}



  def maybe_validate_result(true, result) do

    raise("Task returned #{inspect(result)} but store_result requires a map")

  end

end