Packages
exq
0.8.1
0.23.0
0.22.0
0.21.0
0.20.0
0.19.0
0.18.0
0.17.0
0.16.2
0.16.1
0.16.0
0.15.0
0.14.0
0.13.5
0.13.4
0.13.3
0.13.2
0.13.1
0.13.0
0.12.2
0.12.1
0.12.0
0.11.0
0.10.1
0.10.0
0.9.1
0.9.0
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.3
0.7.2
0.7.1
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.3
0.2.2
0.2.1
0.2.0
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.2
Exq is a job processing library compatible with Resque / Sidekiq for the Elixir language.
Current section
Files
Jump to
Current section
Files
lib/exq/enqueuer/server.ex
defmodule Exq.Enqueuer.Server do
@moduledoc """
The Enqueuer is responsible for enqueueing jobs into Redis. It can
either be called directly by the client, or instantiated as a standalone process.
It supports enqueuing immediate jobs, or scheduling jobs in the future.
## Initialization:
* `:name` - Name of target registered process
* `:namespace` - Redis namespace to store all data under. Defaults to "exq".
* `:queues` - Array of currently active queues (TODO: Remove, I suspect it's not needed).
* `:redis` - pid of Redis process.
* `:scheduler_poll_timeout` - How often to poll Redis for scheduled / retry jobs.
"""
require Logger
alias Exq.Support.Config
alias Exq.Redis.JobQueue
use GenServer
defmodule State do
defstruct redis: nil, namespace: nil
end
def start_link(opts \\ []) do
GenServer.start_link(__MODULE__, opts, name: server_name(opts[:name]))
end
##===========================================================
## gen server callbacks
##===========================================================
def init(opts) do
{:ok, %State{redis: opts[:redis], namespace: opts[:namespace]}}
end
def handle_cast({:enqueue, from, queue, worker, args}, state) do
response = JobQueue.enqueue(state.redis, state.namespace, queue, worker, args)
GenServer.reply(from, response)
{:noreply, state}
end
def handle_cast({:enqueue_at, from, queue, time, worker, args}, state) do
response = JobQueue.enqueue_at(state.redis, state.namespace, queue, time, worker, args)
GenServer.reply(from, response)
{:noreply, state}
end
def handle_cast({:enqueue_in, from, queue, offset, worker, args}, state) do
response = JobQueue.enqueue_in(state.redis, state.namespace, queue, offset, worker, args)
GenServer.reply(from, response)
{:noreply, state}
end
def handle_call({:enqueue, queue, worker, args}, _from, state) do
response = JobQueue.enqueue(state.redis, state.namespace, queue, worker, args)
{:reply, response, state}
end
def handle_call({:enqueue_at, queue, time, worker, args}, _from, state) do
response = JobQueue.enqueue_at(state.redis, state.namespace, queue, time, worker, args)
{:reply, response, state}
end
def handle_call({:enqueue_in, queue, offset, worker, args}, _from, state) do
response = JobQueue.enqueue_in(state.redis, state.namespace, queue, offset, worker, args)
{:reply, response, state}
end
def terminate(_reason, _state) do
:ok
end
# Internal Functions
def server_name(name) do
name = name || Config.get(:name)
"#{name}.Enqueuer" |> String.to_atom
end
end