Packages

honeydew

1.2.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 sources ecto sql cockroach.ex
Raw

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

if Code.ensure_loaded?(Ecto) do



  defmodule Honeydew.EctoSource.SQL.Cockroach do

    alias Honeydew.EctoSource

    alias Honeydew.EctoSource.SQL



    @behaviour SQL



    @impl true

    def integer_type do

      :bigint

    end



    @impl true

    def ready do

      "CAST('#{SQL.far_in_the_past()}' AS TIMESTAMP)"

      |> time_in_msecs

      |> msecs_ago

    end



    @impl true

    def reserve(state) do

      "UPDATE #{state.table}

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

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

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

      LIMIT 1

      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(IF(#{state.lock_field} IS NOT NULL, 1, NULL)) AS count,

              COUNT(IF(#{state.lock_field} >= #{msecs_ago(state.stale_timeout)}, 1, NULL)) AS in_progress,

              COUNT(IF(#{state.lock_field} = #{EctoSource.abandoned()}, 1, NULL)) 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

      time_in_msecs("NOW()")

    end



    defp time_in_msecs(time) do

      "(EXTRACT('millisecond', (#{time})) + CAST((#{time}) AS INT) * 1000)"

    end

  end



end