Initial commit

Author: Cantido

.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

.gitignore

@@ -0,0 +1,24 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+queutils-*.tar
+

README.md

@@ -0,0 +1,20 @@
+# Queutils
+
+Handy little queues and producers to make using `GenStage` a breeze.
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `queutils` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:queutils, "~> 0.1.0"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at [https://hexdocs.pm/queutils](https://hexdocs.pm/queutils).

lib/queutils.ex

@@ -0,0 +1,5 @@
+defmodule Queutils do
+  @moduledoc """
+  Handy little queues and producers.
+  """
+end

lib/queutils/blocking_producer.ex

@@ -0,0 +1,127 @@
+defmodule Queutils.BlockingProducer do
+  use GenStage
+
+  @moduledoc """
+  A `GenStage` producer that acts as a blocking queue, with a fixed length.
+  Blocks any time `Queutils.BlockingProducer.push/2` is called when the queue
+  is at its maximum length.
+
+  This can be used as an entry-point to a `GenStage` pipeline, since the
+  max queue length provides for back-pressure.
+  You can even set the queue's length to zero in order to block all pushes
+  until demand comes in.
+
+
+  ## Usage
+
+  Add it to your application supervisor's `start/2` function like this:
+
+      def start(_type, _args) do
+        children = [
+          ...
+          {Queutils.BlockingProducer, name: MessageProducer, max_length: 10_000},
+          ...
+        ]
+
+        opts = [strategy: :one_for_one, name: MyApplication.Supervisor]
+        Supervisor.start_link(children, opts)
+      end
+
+  Then, you can push messages to the queue like this:
+
+      :ok = Queutils.BlockingProducer.push(MessageProducer, :my_message)
+
+
+  Broadway users be forewared! A Broadway module needs to start its producer itself,
+  so it's not possible to customize the process ID a la the `:name` option documented below.
+  If you're in that boat, you should use a `Queutils.BlockingQueue` along with
+  a `Queutils.BlockingQueueProducer`, so you can customize your reference to your `BlockingQueue`.
+
+  ## Options
+
+    - `:name` - the ID of the queue. This will be the first argument to the `push/2` function.
+    - `:max_length` - The maximum number of messages that this process will store until it starts blocking. Default is 1,000.
+    - `:dispatcher` - The `GenStage` dispatcher that this producer should use. Default is `GenStage.DemandDispatcher`.
+  """
+
+  def start_link(opts) do
+    case Keyword.fetch(opts, :name) do
+      {:ok, name} -> GenStage.start_link(__MODULE__, opts, name: name)
+      :error -> GenStage.start_link(__MODULE__, opts)
+    end
+  end
+
+  def child_spec(opts) do
+    %{
+      id: Keyword.get(opts, :name, Queutils.BlockingProducer),
+      start: {__MODULE__, :start_link, [opts]},
+      type: :supervisor
+    }
+  end
+
+  def push(queue, msg) do
+    GenStage.call(queue, {:push, msg})
+  end
+
+  @impl true
+  def init(opts) do
+    dispatcher = Keyword.get(opts, :dispatcher, GenStage.DemandDispatcher)
+    max_length = Keyword.get(opts, :max_length, 1_000)
+    unless max_length >= 0 do
+      raise "Invalid argument :max_length. Must be an integer zero or greater, but was #{inspect max_length}"
+    end
+    {:producer, %{queue: [], waiting: [], demand: 0, max_length: max_length}, dispatcher: dispatcher}
+  end
+
+  @impl true
+  def handle_call({:push, msg}, from, state) do
+    :ok = validate_state(state)
+    cond do
+      state.demand > 0 ->
+        remaining_demand = state.demand - 1
+        {:reply, :ok, [msg], %{state | demand: remaining_demand}}
+      Enum.count(state.queue) >= state.max_length ->
+        waiting = state.waiting ++ [{from, msg}]
+        {:noreply, [], %{state | waiting: waiting}}
+      true ->
+        queue = state.queue ++ [msg]
+        {:reply, :ok, [], %{state | queue: queue}}
+    end
+  end
+
+  @impl true
+  def handle_demand(demand, state) do
+    :ok = validate_state(state)
+
+    total_demand = demand + state.demand
+
+    {popped, remaining} = Enum.split(state.queue, total_demand)
+    {popped_waiters, still_waiting} = Enum.split(state.waiting, total_demand)
+
+    msgs_from_waiters = Enum.map(popped_waiters, fn {from, msg} ->
+      GenStage.reply(from, :ok)
+      msg
+    end)
+
+    queue = remaining ++ msgs_from_waiters
+    remaining_demand = total_demand - Enum.count(queue)
+
+    {:noreply, popped, %{state | demand: remaining_demand, queue: queue, waiting: still_waiting}}
+  end
+
+  defp validate_state(state) do
+    # If we have a non-zero demand, it is assumed that we will not have
+    # anyone waiting and that the queue is empty, since we would have
+    # popped off all those messages before building up any demand.
+
+    cond do
+      state.demand < 0 ->
+        raise "Incorrect state: BlockingProducer has a negative demand (demand is #{inspect state.demand})."
+      state.demand > 0 && not Enum.empty?(state.queue) ->
+        raise "Incorrect state: BlockingProducer has demand but also has items in its queue."
+      state.demand > 0 && not Enum.empty?(state.waiting) ->
+        raise "Incorrect state: BlockingProducer has demand but also has processes waiting to insert."
+      true -> :ok
+    end
+  end
+end

lib/queutils/blocking_queue.ex

@@ -0,0 +1,90 @@
+defmodule Queutils.BlockingQueue do
+  use GenServer
+
+  @moduledoc """
+  A queue with a fixed length that blocks on `pop/1` if the queue is full.
+
+  ## Usage
+
+  Add it to your application supervisor's `start/2` function, after the queue it pulls from, like this:
+
+      def start(_type, _args) do
+        children = [
+          ...
+          {Queutils.BlockingQueue, name: MessageQueue, max_length: 10_000},
+          ...
+        ]
+
+        opts = [strategy: :one_for_one, name: MyApplication.Supervisor]
+        Supervisor.start_link(children, opts)
+      end
+
+  Then you can push and pop from the queue like this:
+
+      :ok = Queutils.Blockingqueue.push(MessageQueue, :my_message)
+      [:my_message] = Queutils.Blockingqueue.pop(MessageQueue, 1)
+
+  ## Options
+
+    - `:name` - the ID of the queue. This will be the first argument to the `push/2` function. Default is `BlockingQueue`.
+    - `:max_length` - The maximum number of messages that this process will store until it starts blocking. Default is 1,000.
+  """
+
+  def start_link(opts) do
+    name = Keyword.get(opts, :name)
+    GenServer.start_link(__MODULE__, opts, name: name)
+  end
+
+  def child_spec(opts) do
+    %{
+      id: Keyword.get(opts, :name, BlockingQueue),
+      start: {__MODULE__, :start_link, [opts]},
+      type: :supervisor
+    }
+  end
+
+  def init(opts) do
+    max_length = Keyword.get(opts, :max_length, 1_000)
+    {:ok, %{max_length: max_length, queue: [], waiting: []}}
+  end
+
+  def push(queue, msg) do
+    GenServer.call(queue, {:push, msg})
+  end
+
+  def pop(queue, count) do
+    GenServer.call(queue, {:pop, count})
+  end
+
+  def length(queue) do
+    GenServer.call(queue, :length)
+  end
+
+  def handle_call(:length, _from, state) do
+    {:reply, Enum.count(state.queue), state}
+  end
+
+  def handle_call({:push, msg}, from, state) do
+    if Enum.count(state.queue) >= state.max_length do
+      waiting = state.waiting ++ [{from, msg}]
+      {:noreply, %{state | waiting: waiting}}
+    else
+      queue = state.queue ++ [msg]
+      {:reply, :ok, %{state | queue: queue}}
+    end
+  end
+
+  def handle_call({:pop, count}, _from, state) do
+    {popped, remaining} = Enum.split(state.queue, count)
+    {popped_waiters, still_waiting} = Enum.split(state.waiting, count)
+
+    msgs_from_waiters = Enum.map(popped_waiters, fn {from, msg} ->
+      GenServer.reply(from, :ok)
+      msg
+    end)
+
+    queue = remaining ++ msgs_from_waiters
+
+    {:reply, popped, %{state | queue: queue, waiting: still_waiting}}
+  end
+end

lib/queutils/blocking_queue_producer.ex

@@ -0,0 +1,76 @@
+defmodule Queutils.BlockingQueueProducer do
+  use GenStage
+  require Logger
+
+  @moduledoc """
+  A `GenStage` producer that polls a `Queutils.BlockingQueue` at a fixed interval,
+  emitting any events on the queue.
+
+  ## Usage
+
+  Add it to your application supervisor's `start/2` function, after the queue it pulls from, like this:
+
+      def start(_type, _args) do
+        children = [
+          ...
+          {Queutils.BlockingQueue, name: MessageQueue, max_length: 10_000},
+          {Queutils.BlockingQueueProducer, name: MessageProducer},
+          ...
+        ]
+
+        opts = [strategy: :one_for_one, name: MyApplication.Supervisor]
+        Supervisor.start_link(children, opts)
+      end
+
+  The subscribe a consumer to it, like any other `GenStage` producer.
+
+      def init(_opts) do
+        {:consumer, :my_consumer_state, [subscribe_to: MessageProducer]}
+      end
+
+  ## Options
+
+    - `:name` - the ID of the queue. This will be the first argument to the `push/2` function. Default is `BlockingProducer`.
+    - `:max_length` - The maximum number of messages that this process will store until it starts blocking. Default is 1,000.
+    - `:dispatcher` - The `GenStage` dispatcher that this producer should use. Default is `GenStage.DemandDispatcher`.
+  """
+
+  def start_link(opts) do
+    name = Keyword.get(opts, :name, BlockingQueueProducer)
+    GenStage.start_link(__MODULE__, opts, name: name)
+  end
+
+  def child_spec(opts) do
+    %{
+      id: Keyword.get(opts, :name, BlockingQueueProducer),
+      start: {__MODULE__, :start_link, [opts]},
+      type: :supervisor
+    }
+  end
+
+  @impl true
+  def init(opts) do
+    poll_interval = Keyword.get(opts, :poll_interval, 250)
+    dispatcher = Keyword.get(opts, :dispatcher, GenStage.DemandDispatcher)
+    queue = Keyword.get(opts, :queue, BlockingQueue)
+    Process.send_after(self(), :poll, poll_interval)
+    {:producer, %{queue: queue, demand: 0, poll_interval: poll_interval}, dispatcher: dispatcher}
+  end
+
+  @impl true
+  def handle_info(:poll, state) do
+    events = Queutils.BlockingQueue.pop(state.queue, state.demand)
+    remaining_demand = state.demand - Enum.count(events)
+
+    Process.send_after(self(), :poll, state.poll_interval)
+    {:noreply, events, %{state | demand: remaining_demand}}
+  end
+
+  @impl true
+  def handle_demand(demand, state) do
+    total_demand = demand + state.demand
+    events = Queutils.BlockingQueue.pop(state.queue, total_demand)
+    remaining_demand = total_demand - Enum.count(events)
+    {:noreply, events, %{state | demand: remaining_demand}}
+  end
+end

mix.exs

@@ -0,0 +1,42 @@
+defmodule Queutils.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :queutils,
+      version: "1.0.0",
+      elixir: "~> 1.9",
+      start_permanent: Mix.env() == :prod,
+      deps: deps(),
+      description: "Handy little queues and producers.",
+      source_url: "https://github.com/cantido/queutils",
+      homepage_url: "https://github.com/cantido/queutils",
+      package: [
+        maintainers: ["Rosa Richter"],
+        licenses: ["GPL-3.0-or-later"],
+        links: %{"GitHub" => "https://github.com/cantido/queutils"},
+      ],
+      docs: [
+        main: "Queutils",
+        source_url: "https://github.com/cantido/queutils",
+        extras: [
+          "README.md"
+        ]
+      ]
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger]
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:gen_stage, "~> 1.0"}
+    ]
+  end
+end

mix.lock

@@ -0,0 +1,3 @@
+%{
+  "gen_stage": {:hex, :gen_stage, "1.0.0", "51c8ae56ff54f9a2a604ca583798c210ad245f415115453b773b621c49776df5", [:mix], [], "hexpm"},
+}