Code.require_file "test_helper.exs", __DIR__ defmodule ExqTest do use ExUnit.Case import ExqTestUtil defmodule PerformWorker do def perform do send :exqtest, {:worked} end end defmodule PerformArgWorker do def perform(arg) do send :exqtest, {:worked, arg} end end defmodule CustomMethodWorker do def simple_perform do end end defmodule MissingMethodWorker do end defmodule FailWorker do def failure_perform do :num + 1 send :exqtest, {:worked} end end setup do TestRedis.start on_exit fn -> wait TestRedis.stop end :ok end test "start using registered name" do {:ok, exq_sup} = Exq.start_link([port: 6555, name: :custom_manager, namespace: "test"]) assert_exq_up(:custom_manager) stop_process(exq_sup) end test "start multiple exq instances using registered name" do {:ok, sup1} = Exq.start_link([port: 6555, name: :custom_manager1, namespace: "test"]) assert_exq_up(:custom_manager1) {:ok, sup2} = Exq.start_link([port: 6555, name: :custom_manager2, namespace: "test"]) assert_exq_up(:custom_manager2) stop_process(sup1) stop_process(sup2) end test "enqueue and run job" do Process.register(self, :exqtest) {:ok, sup} = Exq.start_link([name: :exq_t, port: 6555, namespace: "test"]) {:ok, _} = Exq.enqueue(:exq_t, "default", "ExqTest.PerformWorker", []) wait assert_received {:worked} stop_process(sup) end test "enqueue with separate enqueuer" do Process.register(self, :exqtest) {:ok, exq_sup} = Exq.start_link([name: :exq_t, port: 6555, namespace: "test"]) {:ok, enq_sup} = Exq.Enqueuer.start_link([name: :exq_e, port: 6555, namespace: "test"]) {:ok, _} = Exq.Enqueuer.enqueue(:exq_e, "default", "ExqTest.PerformWorker", []) wait_long assert_received {:worked} stop_process(exq_sup) stop_process(enq_sup) end test "run jobs on multiple queues" do Process.register(self, :exqtest) {:ok, sup} = Exq.start_link([name: :exq_t, port: 6555, namespace: "test", queues: ["q1", "q2"]]) {:ok, _} = Exq.enqueue(:exq_t, "q1", "ExqTest.PerformArgWorker", [1]) {:ok, _} = Exq.enqueue(:exq_t, "q2", "ExqTest.PerformArgWorker", [2]) wait_long assert_received {:worked, 1} assert_received {:worked, 2} stop_process(sup) end test "record processed jobs" do {:ok, sup} = Exq.start_link([name: :exq_t, port: 6555, namespace: "test"]) state = :sys.get_state(:exq_t) {:ok, _} = Exq.enqueue(:exq_t, "default", "ExqTest.CustomMethodWorker/simple_perform", []) wait {:ok, count} = TestStats.processed_count(state.redis, "test") assert count == "1" {:ok, _} = Exq.enqueue(:exq_t, "default", "ExqTest.CustomMethodWorker/simple_perform", []) wait_long {:ok, count} = TestStats.processed_count(state.redis, "test") assert count == "2" stop_process(sup) end test "record failed jobs" do {:ok, sup} = Exq.start_link([name: :exq_t, port: 6555, namespace: "test"]) state = :sys.get_state(:exq_t) {:ok, _} = Exq.enqueue(:exq_t, "default", "ExqTest.MissingMethodWorker/fail", []) wait_long {:ok, count} = TestStats.failed_count(state.redis, "test") assert count == "1" {:ok, _} = Exq.enqueue(:exq_t, "default", "ExqTest.MissingWorker", []) wait_long {:ok, count} = TestStats.failed_count(state.redis, "test") assert count == "2" {:ok, jid} = Exq.enqueue(:exq_t, "default", "ExqTest.FailWorker/failure_perform", []) # if we kill Exq too fast we dont record the failure because exq is gone wait_long {:ok, enq_sup} = Exq.Enqueuer.start_link([name: :exq_e, port: 6555, namespace: "test"]) # Find the job in the processed queue {:ok, _, _} = Exq.Api.find_failed(:exq_e, jid) wait_long stop_process(sup) stop_process(enq_sup) end end