Packages

queuetopia

0.6.0
2.7.2 2.7.1 2.7.0 2.6.1 2.6.0 2.5.3 2.5.2 2.5.1 2.5.0 2.4.4 2.4.3 2.4.2 2.4.1 2.4.0 2.3.5 2.3.3 2.3.2 2.3.1 2.3.0 2.2.0 2.1.3 2.1.2 2.1.1 2.1.0 2.0.1 2.0.0 1.7.0 1.6.0 1.5.1 1.5.0 1.4.0 1.3.1 1.3.0 1.2.0 1.1.3 1.1.2 1.1.1 1.1.0 1.0.0 0.6.3 0.6.2 0.6.1 0.6.0 0.5.1 0.5.0 0.3.0 0.2.0 0.1.2

Persistent blocking job queue

HexDocs

Current section

Files

Jump to
Readme
48 Versions
3 Dependencies
Dependants
Files
Activity
queuetopia lib queuetopia locks locks.ex
Raw

lib/queuetopia/locks/locks.ex 

defmodule Queuetopia.Locks do

  import Ecto.Query

  alias Queuetopia.Locks.Lock



  @lock_security_retention 1_000



  @spec lock_queue(module(), binary(), binary(), integer()) :: {:error, :locked} | {:ok, Lock.t()}

  def lock_queue(repo, scope, queue, timeout)

      when is_binary(queue) and is_integer(timeout) do

    utc_now = DateTime.utc_now()

    lock_retention = timeout + @lock_security_retention



    %Lock{}

    |> Lock.changeset(%{

      scope: scope,

      queue: queue,

      locked_at: utc_now,

      locked_by_node: Kernel.inspect(Node.self()),

      locked_until: DateTime.add(utc_now, lock_retention, :millisecond)

    })

    |> repo.insert()

    |> case do

      {:ok, %Lock{} = lock} -> {:ok, lock}

      {:error, _changeset} -> {:error, :locked}

    end

  end



  @spec release_expired_locks(module(), binary()) :: any()

  def release_expired_locks(repo, scope) do

    utc_now = DateTime.utc_now()



    Lock

    |> where([lock], lock.scope == ^scope)

    |> where([lock], lock.locked_until <= ^utc_now)

    |> repo.delete_all()

  end



  @spec unlock_queue(module(), binary(), binary()) :: any

  def unlock_queue(repo, scope, queue) do

    Lock

    |> where([lock], lock.scope == ^scope)

    |> where([lock], lock.queue == ^queue)

    |> repo.delete_all()

  end

end