Puller

Puller is a library for Elixir that provides a ready-to-use, configurable architecture for building distributed queues with a pull-based task processing model. It is ideal for systems where workers self-regulate their workload, pulling tasks as they become ready.

Why Puller

Puller eliminates the need to write boilerplate code for GenServer, Supervisor, and Registry. Instead, you define the worker logic and queue, and the library automatically builds a supervision tree that provides:

• Round-robin task distribution across storage.
• Automatic process registration via Registry.
• A pull model where workers request their next task (callback).
• Scalability by configuring the number of workers and queues.

Features

• Pull-based processing: workers pick up tasks themselves when they are ready.
• Distributed queue: the queue is distributed across multiple processes for parallelism..
• Round-robin routing: tasks are distributed evenly between repositories.
• Registry integration: all processes are automatically registered for easy access.
• Supervision tree: full tree with restart capability.
• Telemetry-ready: built-in events for monitoring.
• Configurable: setting the number of workers and queues via application configuration.
• Glory Elixir: minimum boilerplate, just use in your modules.

Installation

Add puller to your list of dependencies in mix.exs:

def deps do
  [
    {:puller, git: "https://github.com/dvolkow/puller.git", branch: "main"}
  ]
end

Quick Start

Generate code with mix.task

mix puller.gen.queue MyApp.Queue
mix puller.gen.worker MyApp.Worker

Configure a Queue

# lib/my_app/queue.ex
defmodule MyApp.Queue do
  use Puller.Queue,
    queue: __MODULE__,
    size: 2  # size of parallel queue for your data
end

Define a Worker

# lib/my_app/worker.ex
defmodule MyApp.Worker do
  use Puller.Worker,
    queue: MyApp.Queue,
    size: 4  # size of parallel workers

  @impl true
  def callback() do
    # here are your logic:
    :ok
  end
end

Add to application supervisor

# lib/my_app/application.ex
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      MyApp.Queue.Supervisor,
      MyApp.Worker.Supervisor
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Or use helper Puller.children:

# lib/my_app/application.ex
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = Puller.children(
      queues: [MyApp.Queue],
      workers: [MyApp.Worker],
    )

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Configuration

config :my_app,
  queues_num: 2, # default queues num
  workers_num: 4 # default workers num

Example of pipeline

1. Putting a task: MyApp.Queue.put(data) → Router picks a storage via round-robin → Storage stores the task.

2. Getting a task: Worker calls MyApp.Queue.get() → Router forwards to a storage → Storage returns the next task.

3. Processing: Worker processes the task when calls callback/0.

Supervision tree

YourApp.Supervisor
├── MyApp.Queue.Supervisor (one_for_one)
│   ├── MyApp.Queue.Router (GenServer) - round-robin distributor
│   ├── Registry (MyApp.Queue.Registry)
│   ├── MyApp.Queue-0 (GenServer) - queue storage
│   ├── MyApp.Queue-1 (GenServer)
│   └── MyApp.Queue-2 (GenServer)
│   ...
└── MyApp.Worker.Supervisor (one_for_one)
    ├── Registry (MyApp.Worker.Registry)
    ├── MyApp.Worker-0 (GenServer) - pull-based worker
    ├── MyApp.Worker-1 (GenServer)
    └── MyApp.Worker-2 (GenServer)