Merge pull request #54 from shinyscorpion/feature/reject_without_retry

Reject job without retry option

README.md

@@ -330,6 +330,12 @@ You can also change the retry_interval by the number of failures.
 
 If a job fails more than `max_retry` times, the payload is sent to `jobs.[job_name].rejected` queue.
 
+#### Immediately Reject
+
+TaskBunny can mark a job as rejected without retrying when `perform` returns `:reject` or `{:reject, something}`
+
+In this case any `max_retry` config is ignored.
+
 #### Timeout
 
 By default, jobs timeout after 2 minutes.

lib/task_bunny/worker.ex

@@ -243,15 +243,19 @@ defmodule TaskBunny.Worker do
 
     FailureBackend.report_job_error(job_error)
 
-    if failed_count <= job.max_retry() do
-      retry_message(job, state, new_body, meta, failed_count)
-      {:noreply, update_job_stats(state, :failed)}
-    else
+    if reject?(job, failed_count, job_error) do
       reject_message(state, new_body, meta)
       {:noreply, update_job_stats(state, :rejected)}
+    else
+      retry_message(job, state, new_body, meta, failed_count)
+      {:noreply, update_job_stats(state, :failed)}
     end
   end
 
+  defp reject?(_, _, %{return_value: :reject}), do: true
+  defp reject?(_, _, %{return_value: {:reject, _}}), do: true
+  defp reject?(job, failed_count, _), do: failed_count > job.max_retry()
+
   @spec retry_message(atom, Worker.t(), any, any, integer) :: :ok
   defp retry_message(job, state, body, meta, failed_count) do
     retry_queue = Queue.retry_queue(state.queue)

test/support/job_test_helper.ex

@@ -15,10 +15,10 @@ defmodule TaskBunny.JobTestHelper do
 
       if payload["sleep"], do: :timer.sleep(payload["sleep"])
 
-      if payload["fail"] do
-        :error
-      else
-        :ok
+      cond do
+        payload["reject"] -> :reject
+        payload["fail"] -> :error
+        true -> :ok
       end
     end
 

test/task_bunny/worker_test.exs

@@ -130,6 +130,26 @@ defmodule TaskBunny.WorkerTest do
 
       GenServer.stop(worker)
     end
+
+    test "sends to rejected queue if return_value is :reject" do
+      worker = start_worker()
+      [main, retry, rejected, _scheduled] = all_queues()
+      payload = %{"reject" => true}
+
+      TestJob.enqueue(payload, queue: @queue)
+      JobTestHelper.wait_for_perform()
+
+      conn = Connection.get_connection!()
+      %{message_count: main_count} = Queue.state(conn, main)
+      %{message_count: retry_count} = Queue.state(conn, retry)
+      %{message_count: rejected_count} = Queue.state(conn, rejected)
+
+      assert main_count == 0
+      assert retry_count == 0
+      assert rejected_count == 1
+
+      GenServer.stop(worker)
+    end
   end
 
   describe "delay" do