Creating a Looping GenServer

This week’s post is about a 5 minute read.

The video is a little over 2 minutes long.

GenServers are very easy to setup, but it’s not immediately obvious how to implement a looping server that does work on its own.

For example, let’s say we’re writing a job queue and we want our GenServer worker to respond to jobs placed into Redis. We want our server to pop jobs from Redis, process the job, and then start over again for the next incoming job.

The key to setting this all up is to remember that we can use handle_info as a swiss army knife for “out of band” messages and that we can kick these off from the initializer:

defmodule NaivePoller do
  use GenServer

  def start_link(queue_name) do
    GenServer.start_link __MODULE__, queue_name, name: __MODULE__
  end

  def init(queue_name) do
    {:ok, conn} = Redix.start_link

    schedule_poller()

    {:ok, %{conn: conn, queue: queue_name}}
  end

  def handle_info(:poll, state = %{conn: conn, queue: queue_name}) do
    job = dequeue_job(conn, queue_name)

    SomeWorker.handle(job)

    schedule_poller()

    {:noreply, state}
  end

  defp schedule_poller() do
    send self(), :poll
  end

  defp dequeue_job(conn, queue_name) do
    Redix.command!(conn, ["BRPOP", queue_name, 0], timeout: :infinity)
  end
end

Let’s walk through this code.

In init we use Redix to initialize our Redis client. And then on the following line we call schedule_poller which shoves the message :poll into our mailbox.

The initializer completes normally, setting up a simple map for the server state which wraps the Redis connection along with the name of the queue we’re interested in:

def init(queue_name) do
  # setup up our Redis client
  {:ok, conn} = Redix.start_link

  # shove :poll into our mailbox
  schedule_poller() # => send self(), :poll

  # return our server state
  {:ok, %{conn: conn, queue: queue_name}}
end

handle_info(:poll, state) is then immediately invoked to deal with the :poll message we sent ourselves in init.

We use this as an opportunity to call dequeue_job which makes a blocking call to Redis via BRPOP (ignore that there are better patterns for this, for now).

def handle_info(:poll, state = %{conn: conn, queue: queue_name}) do
  # dequeue the job
  job = dequeue_job(conn, queue_name)

  # handle the job
  SomeWorker.handle(job)

  # set ourselves up to recur
  schedule_poller() # => send self(), :poll

  # return from handle_info
  {:noreply, state}
end

Basically we’re just waiting around for a job to come in.

Once a job appears in Redis, dequeue_job will pop it off so that the hypothetical worker SomeWorker can process it.

After the job has been processed we make another call to schedule_poller, which shoves :poll back into the mailbox causing the whole process to start over again.

So, in essence, our GenServer is recurring to get work done. Pretty cool.

Here is a quick video that shows how the code runs:

Creating a Looping GenServer from Jim Whiteman on Vimeo.