Packages

Simple and reliable background job processing library for Elixir. Has persistence, retries, delayed jobs, concurrency limiting, error handling and is heroku friendly.

Current section

Files

Jump to
toniq lib toniq keepalive_persistence.ex
Raw

lib/toniq/keepalive_persistence.ex

defmodule Toniq.KeepalivePersistence do
def register_vm(identifier) do
redis_query(["SADD", registered_vms_key, identifier])
end
def update_alive_key(identifier, keepalive_expiration) do
# Logger.log(:debug, "Updating keepalive for #{state.identifier} #{inspect(debug_info)}")
redis_query(["PSETEX", alive_key(identifier), keepalive_expiration, debug_info])
end
def registered_vms do
redis_query(["SMEMBERS", registered_vms_key])
end
def alive?(identifier) do
alive_vm_debug_info(identifier) != :undefined
end
def alive_vm_debug_info(identifier) do
redis_query(["GET", alive_key(identifier)])
end
def takeover_jobs(from_identifier, to_identifier) do
redis
|> Exredis.query_pipe([
# Begin transaction
["MULTI"],
# Copy orphaned jobs to incoming jobs
#
# We copy jobs to incoming jobs so that they will be
# enqueued and run in this vm. Data in redis is just
# a backup, we only poll for incoming jobs.
[
"SUNIONSTORE",
incoming_jobs_key(to_identifier),
jobs_key(from_identifier),
incoming_jobs_key(from_identifier),
incoming_jobs_key(to_identifier)
],
# Move failed jobs
[
"SUNIONSTORE",
failed_jobs_key(to_identifier),
failed_jobs_key(from_identifier),
failed_jobs_key(to_identifier),
],
# Remove orphaned job lists
[
"DEL",
jobs_key(from_identifier),
failed_jobs_key(from_identifier),
incoming_jobs_key(from_identifier)
],
# Deregister missing vm
[
"SREM",
registered_vms_key,
from_identifier
],
# Execute transaction
["EXEC"]
])
end
# Added so we could use it to default scope further out when we want to allow custom persistance scopes in testing.
def default_scope, do: Application.get_env(:toniq, :redis_key_prefix)
defp incoming_jobs_key(identifier) do
Toniq.JobPersistence.incoming_jobs_key(identifier)
end
defp jobs_key(identifier) do
Toniq.JobPersistence.jobs_key(identifier)
end
defp failed_jobs_key(identifier) do
Toniq.JobPersistence.failed_jobs_key(identifier)
end
defp redis_query(query) do
redis |> Exredis.query(query)
end
defp redis do
:toniq_redis
|> Process.whereis
end
# This is not a API any production code should rely upon, but could be useful
# info when debugging or to verify things in tests.
defp debug_info, do: %{ system_pid: System.get_pid, last_updated_at: :os.system_time }
defp alive_key(identifier), do: "#{default_scope}:#{identifier}:alive"
defp registered_vms_key, do: "#{default_scope}:registered_vms"
end