Packages

step_flow

0.0.16
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

HexDocs

Current section

Files

Jump to
Readme
128 Versions
15 Dependencies
Dependants
Files
Activity
step_flow lib step_flow controllers workflow_events_controller.ex
Raw

lib/step_flow/controllers/workflow_events_controller.ex 

defmodule StepFlow.WorkflowEventsController do

  use StepFlow, :controller

  require Logger



  alias StepFlow.Amqp.CommonEmitter

  alias StepFlow.Jobs

  alias StepFlow.Jobs.Status

  alias StepFlow.Workflows



  action_fallback(StepFlow.FallbackController)



  def handle(conn, %{"id" => id} = params) do

    workflow = Workflows.get_workflow!(id)



    case params do

      %{"event" => "abort"} ->

        workflow.steps

        |> skip_remaining_steps(workflow)



        topic = "update_workflow_" <> Integer.to_string(workflow.id)

        StepFlow.Notification.send(topic, %{workflow_id: workflow.id})



        conn

        |> put_status(:ok)

        |> json(%{status: "ok"})



      %{"event" => "retry", "job_id" => job_id} ->

        Logger.warn("retry job #{job_id}")



        job = Jobs.get_job_with_status!(job_id)



        last_status = Status.get_last_status(job.status)



        if last_status.state == :error do

          Status.set_job_status(job_id, :retrying)



          params = %{

            job_id: job.id,

            parameters: job.parameters

          }



          case CommonEmitter.publish_json(job.name, job.step_id, params) do

            :ok ->

              StepFlow.Notification.send("retry_job", %{workflow_id: workflow.id, body: params})



              conn

              |> put_status(:ok)

              |> json(%{status: "ok"})



            _ ->

              conn

              |> put_status(:ok)

              |> json(%{status: "error", message: "unable to publish message"})

          end

        else

          send_resp(conn, :forbidden, "illegal operation")

        end



      %{"event" => "delete"} ->

        for job <- workflow.jobs do

          Jobs.delete_job(job)

        end



        Workflows.delete_workflow(workflow)

        StepFlow.Notification.send("delete_workflow", %{workflow_id: workflow.id})



        conn

        |> put_status(:ok)

        |> json(%{status: "ok"})



      _ ->

        send_resp(conn, 422, "event is not supported")

    end

  end



  defp skip_remaining_steps([], _workflow), do: nil



  defp skip_remaining_steps([step | steps], workflow) do

    case step.status do

      :queued -> StepFlow.Step.skip_step(workflow, step)

      :processing -> StepFlow.Step.skip_step_jobs(workflow, step)

      _ -> nil

    end



    skip_remaining_steps(steps, workflow)

  end

end