Packages

honeydew

1.1.0
1.5.0 1.4.6 1.4.5 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.3.0 1.2.8 1.2.7 1.2.6 1.2.5 1.2.4 1.2.3 1.2.2 1.2.1 1.2.0 1.1.6 1.1.5 1.1.4 1.1.3 1.1.2 1.1.1 1.1.0 1.0.4 1.0.3 1.0.2 1.0.1 1.0.0 1.0.0-rc7 1.0.0-rc6 1.0.0-rc5 1.0.0-rc4 1.0.0-rc3 1.0.0-rc2 1.0.0-rc1 0.0.11 0.0.10 0.0.9 0.0.8 0.0.7 0.0.6 0.0.5 0.0.4 0.0.3 0.0.2 0.0.1

Pluggable local/clusterable job queue focused on safety.

HexDocs

Current section

Files

Jump to
Readme
48 Versions
1 Dependency
3 Dependants
Files
Activity
honeydew lib honeydew poll_queue.ex
Raw

lib/honeydew/poll_queue.ex 

defmodule Honeydew.PollQueue do

  require Logger

  alias Honeydew.Job

  alias Honeydew.Queue

  alias Honeydew.Queue.State, as: QueueState



  @behaviour Queue



  @type job :: Job.t()

  @type private :: any()

  @type name :: Honeydew.queue_name()



  @callback init(name, arg :: any()) :: {:ok, private}

  @callback reserve(private) :: {job, private}

  @callback ack(job, private) :: private

  @callback nack(job, private) :: private

  @callback status(private) :: %{:count => number, :in_progress => number, optional(atom) => any}

  @callback cancel(job, private) :: {:ok | {:error, :in_progress | :not_found}, private}



  @callback handle_info(msg :: :timeout | term, state :: private) ::

              {:noreply, new_state}

              | {:noreply, new_state, timeout | :hibernate}

              | {:stop, reason :: term, new_state}

            when new_state: private



  defmodule State do

    defstruct [:queue, :source, :interval]

  end



  @impl true

  def init(queue, [source, args]) do

    interval = Keyword.get(args, :interval, 1_000)



    {:ok, source_state} = source.init(queue, args)

    source = {source, source_state}



    poll(interval)



    {:ok, %State{queue: queue, source: source, interval: interval}}

  end



  @impl true

  def enqueue(job, state) do

    nack(job, state)

    {state, job}

  end



  @impl true

  def reserve(%State{source: {source, source_state}} = state) do

    case source.reserve(source_state) do

      {:empty, source_state} ->

        {:empty, %{state | source: {source, source_state}}}



      {{:value, {id, job}}, source_state} ->

        {%{job | private: id}, %{state | source: {source, source_state}}}

    end

  end



  @impl true

  def ack(job, %State{source: {source, source_state}} = state) do

    %{state | source: {source, source.ack(job, source_state)}}

  end



  @impl true

  def nack(job, %State{source: {source, source_state}} = state) do

    %{state | source: {source, source.nack(job, source_state)}}

  end



  @impl true

  def status(%State{source: {source, source_state}}) do

    source.status(source_state)

  end



  @impl true

  def filter(_state, _function) do

    raise "filter/2 is unsupported for poll queues"

  end



  @impl true

  def cancel(job, %State{source: {source, source_state}} = state) do

    {response, source_state} = source.cancel(job, source_state)

    {response, %{state | source: {source, source_state}}}

  end



  @impl true

  def handle_info(:__poll__, %QueueState{private: %State{interval: interval}} = queue_state) do

    poll(interval)

    {:noreply, Queue.dispatch(queue_state)}

  end



  @impl true

  def handle_info(msg, %QueueState{private: %State{source: {source, _source_state}}} = queue_state) do

    source.handle_info(msg, queue_state)

  end



  defp poll(interval) do

    {:ok, _} = :timer.send_after(interval, :__poll__)

  end

end