Packages

honeydew

1.2.3
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 sources ecto sql postgres.ex
Raw

lib/honeydew/sources/ecto/sql/postgres.ex 

if Code.ensure_loaded?(Ecto) do



  defmodule Honeydew.EctoSource.SQL.Postgres do

    alias Honeydew.EctoSource

    alias Honeydew.EctoSource.SQL



    @behaviour SQL



    @impl true

    def integer_type do

      :bigint

    end



    @impl true

    def ready do

      SQL.far_in_the_past()

      |> time_in_msecs

      |> msecs_ago

    end



    @impl true

    def reserve(state) do

      "UPDATE #{state.table}

      SET #{state.lock_field} = #{now_msecs()}

      WHERE id = (

        SELECT id

        FROM #{state.table}

        WHERE #{state.lock_field} BETWEEN 0 AND #{msecs_ago(state.stale_timeout)}

        ORDER BY #{state.lock_field}, #{state.key_field}

        LIMIT 1

        FOR UPDATE SKIP LOCKED

      )

      RETURNING #{state.key_field}, #{state.private_field}"

    end



    @impl true

    def cancel(state) do

      "UPDATE #{state.table}

      SET #{state.lock_field} = NULL

      WHERE

        id = $1

        AND #{state.lock_field} BETWEEN 0 AND #{msecs_ago(state.stale_timeout)}

      RETURNING #{state.lock_field}"

    end



    @impl true

    def status(state) do

      "SELECT COUNT(CASE WHEN #{state.lock_field} IS NOT NULL THEN 1 ELSE NULL END) AS count,

              COUNT(CASE WHEN #{state.lock_field} >= #{msecs_ago(state.stale_timeout)} THEN 1 ELSE NULL END) AS in_progress,

              COUNT(CASE WHEN #{state.lock_field} = #{EctoSource.abandoned()} THEN 1 ELSE NULL END) AS abandoned

      FROM #{state.table}"

    end



    @impl true

    def filter(state, :abandoned) do

      "SELECT id FROM #{state.table} WHERE #{state.lock_field} = #{EctoSource.abandoned()}"

    end



    defp msecs_ago(msecs) do

      "#{now_msecs()} - #{msecs}"

    end



    defp now_msecs do

      "(CAST(EXTRACT(epoch FROM NOW()) * 1000 AS BIGINT))"

    end



    defp time_in_msecs(time) do

      "(CAST(EXTRACT(epoch from timestamp '#{time}') * 1000 AS BIGINT))"

    end



  end



end