blocking_queue

defmodule BlockingQueue do
  @moduledoc """
  BlockingQueue is a simple queue implemented as a GenServer.  It has a fixed
  maximum length.

  The queue is designed to decouple but limit the latency of a producer and
  consumer.  When pushing to a full queue the `push` operation blocks
  preventing the producer from making progress until the consumer catches up.
  Likewise, when calling `pop` on an empty queue the call blocks until there
  is work to do.

  ## Examples

      {:ok, pid} = BlockingQueue.start_link(5)
      BlockingQueue.push(pid, "Hi")
      BlockingQueue.pop(pid) # should return "Hi"
  """
  use GenServer

  # Can I get this from somewhere?
  @type on_start :: {:ok, pid} | :ignore | {:error, {:already_started, pid} | term}

  @doc """
  Start a queue process with GenServer.start_link/2.

  `n` Is the maximum queue depth.
  """
  @spec start_link(pos_integer()) :: on_start
  def start_link(n), do: GenServer.start_link(__MODULE__, n)
  def init(n), do: {:ok, {n, []}}

  @typep from_t   :: {pid, any}
  @typep state_t  :: {pos_integer(), [any]}
                   | {pos_integer(), [any], :pop, from_t}
                   | {pos_integer(), [any], :push, from_t, any}
  @typep call_t   :: {:push, any}
                   | :pop
  @typep result_t :: {:reply, any, state_t}
                   | {:noreply, state_t}

  @spec handle_call(call_t, from_t, state_t) :: result_t

  def handle_call({:push, item}, waiter, {max, list}) when length(list) > max do
    {:noreply, {max, list, :push, waiter, item}}
  end

  def handle_call({:push, item}, _, {max, list}) do
    {:reply, nil, { max, list ++ [item] }}
  end

  def handle_call({:push, item}, _, {max, [], :pop, from}) do
    GenServer.reply(from, item)
    {:reply, nil, {max, []}}
  end

  def handle_call(:pop, from, {max, []}), do: {:noreply, {max, [], :pop, from}}
  def handle_call(:pop, _, {max, [x | xs]}), do: {:reply, x, {max, xs}}
  def handle_call(:pop, _, {max, [x | xs], :push, waiter, item}) do
    GenServer.reply(waiter, nil)
    {:reply, x, {max, xs ++ [item]}}
  end

  @doc """
  Pushes a new item into the queue.  Blocks if the queue is full.

  `pid` is the process ID of the BlockingQueue server.
  `item` is the value to be pushed into the queue.  This can be anything.
  """
  @spec push(pid, any) :: nil
  def push(pid, item), do: GenServer.call(pid, {:push, item})

  @doc """
  Pops the least recently pushed item from the queue. Blocks if the queue is
  empty until an item is available.

  `pid` is the process ID of the BlockingQueue server.
  """
  @spec pop(pid) :: any
  def pop(pid), do: GenServer.call(pid, :pop)
end