defmodule SPQueue.Test do use ExUnit.Case, async: true doctest SPQueue test "initial state", %{line: line} = _context do pq = start_unique(line) assert SPQueue.head(pq) == {:error, :empty} assert_raise SPQueue.EmptyError, fn -> SPQueue.head!(pq) end assert SPQueue.count(pq) == 0 assert SPQueue.empty?(pq) stop(pq) end test "simple enqueue/dequeue", %{line: line} = _context do pq = start_unique(line) 1..3 |> Enum.each(fn n -> {:ok, %{"msg" => %{"n" => ^n}, "id" => _id, "ts" => _ts}} = SPQueue.enqueue(pq, %{"n" => n}) end) assert SPQueue.count(pq) == 3 {:ok, %{"msg" => %{"n" => 1}, "id" => _id, "ts" => _ts}} = SPQueue.head(pq) 1..3 |> Enum.each(fn n -> {:ok, %{"msg" => %{"n" => ^n}, "id" => _id, "ts" => _ts}} = SPQueue.dequeue(pq) end) assert SPQueue.empty?(pq) stop(pq) end test "simple enqueue/dequeue !", %{line: line} = _context do pq = start_unique(line) 1..3 |> Enum.each(fn n -> SPQueue.enqueue!(pq, %{"n" => n}) end) assert SPQueue.count(pq) == 3 assert SPQueue.head!(pq) == %{"n" => 1} 1..3 |> Enum.each(fn n -> assert SPQueue.dequeue!(pq) |> Map.get("n") == n end) assert SPQueue.empty?(pq) stop(pq) end test "load state", %{line: line} = _context do pq = start_unique(line) 1..3 |> Enum.each(fn n -> SPQueue.enqueue(pq, %{"n" => n}) end) SPQueue.dequeue(pq) assert SPQueue.count(pq) == 2 assert SPQueue.head!(pq) == %{"n" => 2} stop(pq, clean: false) pq = start_unique(line, clean: false) assert SPQueue.count(pq) == 2 assert SPQueue.head!(pq) == %{"n" => 2} stop(pq) end test "multiple segments 32/16", %{line: line} = _context do pq = start_unique(line) 1..32 |> Enum.each(fn n -> SPQueue.enqueue(pq, %{"n" => n}) end) 1..16 |> Enum.each(fn n -> assert SPQueue.dequeue!(pq) |> Map.get("n") == n end) assert SPQueue.head!(pq) == %{"n" => 17} assert SPQueue.count(pq) == 16 stop(pq) end test "multiple segments 40", %{line: line} = _context do pq = start_unique(line) 1..40 |> Enum.each(fn n -> SPQueue.enqueue(pq, %{"n" => n}) end) assert SPQueue.count(pq) == 40 1..40 |> Enum.each(fn n -> assert SPQueue.dequeue!(pq) == %{"n" => n} end) assert SPQueue.empty?(pq) stop(pq) end test "load state multiple segments", %{line: line} = _context do pq = start_unique(line) 1..32 |> Enum.each(fn n -> SPQueue.enqueue(pq, %{"n" => n}) end) 1..16 |> Enum.each(fn n -> assert SPQueue.dequeue!(pq) |> Map.get("n") == n end) assert SPQueue.head!(pq) == %{"n" => 17} assert SPQueue.count(pq) == 16 stop(pq, clean: false) pq = start_unique(line, clean: false) assert SPQueue.head!(pq) == %{"n" => 17} assert SPQueue.count(pq) == 16 stop(pq) end test "options", %{line: line} = _context do queue_name = unique_name_for_test(line) tmp_dir = System.tmp_dir!() number_of_segments = 5 segment_size = 20 maximum_size = number_of_segments * segment_size {:ok, pq} = SPQueue.start( name: String.to_atom(queue_name), base_dir: tmp_dir, number_of_segments: number_of_segments, segment_size: segment_size ) opts = SPQueue.info(pq) assert opts[:count] == 0 assert opts[:maximum_size] == maximum_size assert opts[:name] == String.to_atom(queue_name) assert opts[:queue_base_dir] == Path.join(tmp_dir, queue_name) 1..maximum_size |> Enum.each(fn i -> SPQueue.enqueue!(pq, %{"key" => i}) end) opts = SPQueue.info(pq) assert opts[:count] == maximum_size 1..maximum_size |> Enum.each(fn i -> assert SPQueue.dequeue!(pq) == %{"key" => i} end) opts = SPQueue.info(pq) assert opts[:count] == 0 stop(pq) end test "full", %{line: line} = _context do pq = start_unique(line) max = SPQueue.info(pq)[:maximum_size] 1..max |> Enum.each(fn n -> SPQueue.enqueue!(pq, %{"n" => n}) end) refute(SPQueue.empty?(pq)) assert SPQueue.count(pq) == max assert SPQueue.enqueue(pq, %{"n" => max + 1}) == {:error, :full} assert_raise SPQueue.FullError, fn -> SPQueue.enqueue!(pq, %{"n" => max + 1}) end assert_raise SPQueue.TimedOutError, fn -> SPQueue.enqueue_wait!(pq, %{"n" => max + 1}, timeout: 10, step: 2) end stop(pq) end test "gc", %{line: line} = _context do pq = start_unique(line) 0..4 |> Enum.each(fn round -> 1..40 |> Enum.each(fn i -> {:ok, _} = SPQueue.enqueue(pq, %{"n" => round * 40 + i}) end) assert SPQueue.count(pq) == 40 1..40 |> Enum.each(fn _i -> {:ok, _} = SPQueue.dequeue(pq) end) assert SPQueue.empty?(pq) end) assert Enum.empty?(queue_base_dir(pq) |> File.ls!()) stop(pq) end test "harmonica", %{line: line} = _context do pq = start_unique(line) [{40, 10}, {10, 30}, {20, 10}, {10, 30}, {20, 20}, {15, 5}, {15, 5}, {10, 15}, {10, 25}] |> Enum.reduce({0, 0}, fn {to_enqueue, to_dequeue}, {total_enqueued, total_dequeued} -> total_enqueued_next = total_enqueued + to_enqueue total_enqueued..(total_enqueued_next - 1) |> Enum.each(fn n -> SPQueue.enqueue!(pq, %{"n" => n}) end) total_dequeued_next = total_dequeued + to_dequeue total_dequeued..(total_dequeued_next - 1) |> Enum.each(fn _n -> SPQueue.dequeue!(pq) end) {total_enqueued_next, total_dequeued_next} end) assert Enum.empty?(queue_base_dir(pq) |> File.ls!()) stop(pq) end test "dequeue with id", %{line: line} = _context do pq = start_unique(line) 100..105 |> Enum.each(fn n -> {:ok, _record} = SPQueue.enqueue(pq, %{"code" => n}) end) {:ok, _record} = SPQueue.dequeue(pq) {:ok, %{"id" => id} = _record} = SPQueue.head(pq) {:ok, _record} = SPQueue.dequeue(pq, id: id) {:error, :mismatch} = SPQueue.dequeue(pq, id: id) assert_raise SPQueue.MismatchError, fn -> SPQueue.dequeue!(pq, id: id) end 1..4 |> Enum.each(fn i -> {:ok, _record} = SPQueue.dequeue(pq, id: id + i) end) {:error, :empty} = SPQueue.dequeue(pq) {:error, :empty} = SPQueue.head(pq) stop(pq) end test "to list conversion", %{line: line} = _context do pq = start_unique(line) 1..10 |> Enum.each(fn i -> SPQueue.enqueue(pq, %{"i" => i}) end) assert SPQueue.to_list(pq) == 1..10 |> Enum.map(fn i -> %{"i" => i} end) 11..50 |> Enum.each(fn i -> SPQueue.enqueue(pq, %{"i" => i}) end) assert SPQueue.to_list(pq) == 1..50 |> Enum.map(fn i -> %{"i" => i} end) stop(pq) end test "delegate receives dequeued", %{line: line} = _context do pq = start_unique(line, delegate: self()) SPQueue.enqueue(pq, %{"test" => 123}) assert_receive :enqueued stop(pq) end test "simple worker", %{line: line} = _context do queue_name = String.to_atom(unique_name_for_test(line)) worker_name = String.to_atom("test-worker-#{line}") test_process = self() {:ok, worker} = SPQueueWorker.start( queue_name: queue_name, name: worker_name, handler_function: fn msg -> send(test_process, msg) :ack end ) pq = start_unique(line, delegate: worker) SPQueue.enqueue(pq, %{"test" => 123}) assert_receive %{"test" => 123} stop(pq) SPQueueWorker.stop(worker) end test "drain queue by worker", %{line: line} = _context do queue_name = String.to_atom(unique_name_for_test(line)) worker_name = String.to_atom("test-worker-#{line}") test_process = self() limit = 1000 {:ok, worker} = SPQueueWorker.start( queue_name: queue_name, name: worker_name, handler_function: fn %{"i" => i} = _msg -> if i == limit, do: send(test_process, :done) :ack end ) pq = start_unique(line, delegate: worker) 1..limit |> Enum.each(fn i -> SPQueue.enqueue_wait!(pq, %{"i" => i}) end) assert_receive :done assert SPQueue.empty?(pq) stop(pq) SPQueueWorker.stop(worker) end test "worker periodic interval", %{line: line} = _context do queue_name = String.to_atom(unique_name_for_test(line)) pq = start_unique(line) ~w(x y z) |> Enum.each(fn c -> SPQueue.enqueue!(pq, %{"c" => c}) end) stop(pq, clean: false) worker_name = String.to_atom("test-worker-#{line}") test_process = self() {:ok, worker} = SPQueueWorker.start( queue_name: queue_name, name: worker_name, handler_function: fn %{"c" => c} = _msg -> if c == "z", do: send(test_process, :done) :ack end, periodic_interval: 10 ) pq = start_unique(line, delegate: worker, clean: false) assert_receive :done assert SPQueue.empty?(pq) stop(pq) SPQueueWorker.stop(worker) end # support defp unique_name_for_test(id), do: "test-pq-#{id}" defp start_unique(id, opts \\ [clean: true]) do name = unique_name_for_test(id) Keyword.get(opts, :clean, true) && File.rm_rf!(name) {:ok, pq} = SPQueue.start([name: String.to_atom(name)] ++ opts) pq end defp stop(pq, opts \\ [clean: true]) do Keyword.get(opts, :clean, true) && queue_base_dir(pq) |> File.rm_rf!() SPQueue.stop(pq) end defp queue_base_dir(pq) do SPQueue.info(pq)[:queue_base_dir] end end