Repository: plataformatec/broadway
Pull request #42 (2019-02-14 12:28:31)

Add Broadway.TermStorage


Patch:
diff --git a/lib/broadway/application.ex b/lib/broadway/application.ex
new file mode 100644
index 0000000..eab9096
--- /dev/null
+++ b/lib/broadway/application.ex
@@ -0,0 +1,8 @@
+defmodule Broadway.Application do
+  @moduledoc false
+  use Application
+
+  def start(_type, _args) do
+    Supervisor.start_link([Broadway.TermStorage], strategy: :one_for_one, name: __MODULE__)
+  end
+end
diff --git a/lib/broadway/term_storage.ex b/lib/broadway/term_storage.ex
new file mode 100644
index 0000000..dbd22a8
--- /dev/null
+++ b/lib/broadway/term_storage.ex
@@ -0,0 +1,67 @@
+defmodule Broadway.TermStorage do
+  @moduledoc """
+  A simple term storage to avoid passing large amounts of data between processes.
+
+  If you have a large amount of data and you want to avoid passing it between
+  processes, you can use the `TermStorage`. The `TermStorage` creates a unique
+  reference for it, allowing you pass the reference around instead of the term.
+
+  If the same term is put multiple times, it is stored only once, avoiding
+  generating garbage. However, the stored terms are never removed.
+  A common use case for this feature is in Broadway.Producer, which may need
+  to pass a lot of information to acknowledge messages. With this module, you
+  can store those terms when the producer starts and only pass the reference
+  between messages:
+
+      iex> ref = Broadway.TermStorage.put({:foo, :bar, :baz}) # On init
+      iex> Broadway.TermStorage.get!(ref) # On ack
+      {:foo, :bar, :baz}
+
+  """
+
+  use GenServer
+  @name __MODULE__
+
+  @doc false
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, :ok, name: @name)
+  end
+
+  @doc """
+  Gets a previously stored term.
+  """
+  def get!(ref) when is_reference(ref) do
+    :ets.lookup_element(@name, ref, 2)
+  end
+
+  @doc """
+  Puts a term.
+  """
+  def put(term) do
+    find_by_term(term) || GenServer.call(@name, {:put, term}, :infinity)
+  end
+
+  # Callbacks
+
+  def init(:ok) do
+    ets = :ets.new(@name, [:named_table, :protected, :set, read_concurrency: true])
+    {:ok, ets}
+  end
+
+  def handle_call({:put, term}, _from, state) do
+    if ref = find_by_term(term) do
+      {:reply, ref, state}
+    else
+      ref = make_ref()
+      :ets.insert(@name, {ref, term})
+      {:reply, ref, state}
+    end
+  end
+
+  defp find_by_term(term) do
+    case :ets.match(@name, {:"$1", term}) do
+      [[ref]] -> ref
+      [] -> nil
+    end
+  end
+end
diff --git a/mix.exs b/mix.exs
index 2bf02a2..d6867f3 100644
--- a/mix.exs
+++ b/mix.exs
@@ -19,7 +19,8 @@ defmodule Broadway.MixProject do
 
   def application do
     [
-      extra_applications: [:logger]
+      extra_applications: [:logger],
+      mod: {Broadway.Application, []}
     ]
   end
 


Test patch:
diff --git a/test/broadway/batcher_test.exs b/test/broadway/batcher_test.exs
index ddf7ef3..1110789 100644
--- a/test/broadway/batcher_test.exs
+++ b/test/broadway/batcher_test.exs
@@ -1,5 +1,5 @@
 defmodule Broadway.BatcherTest do
-  use ExUnit.Case
+  use ExUnit.Case, async: true
 
   test "max_demand defaults to batch_size" do
     {_, state, _} =
diff --git a/test/broadway/options_test.exs b/test/broadway/options_test.exs
index 9a6702b..4e0f0fa 100644
--- a/test/broadway/options_test.exs
+++ b/test/broadway/options_test.exs
@@ -1,6 +1,7 @@
 defmodule Broadway.OptionsTest do
-  use ExUnit.Case
+  use ExUnit.Case, async: true
 
+  doctest Broadway.Options
   alias Broadway.Options
 
   test "known options" do
diff --git a/test/broadway/processor_test.exs b/test/broadway/processor_test.exs
index fc8f4f0..ad2e332 100644
--- a/test/broadway/processor_test.exs
+++ b/test/broadway/processor_test.exs
@@ -1,5 +1,5 @@
 defmodule Broadway.ProcessorTest do
-  use ExUnit.Case
+  use ExUnit.Case, async: true
 
   test "set custom min and max demand" do
     {_, state, _} =
diff --git a/test/broadway/producer_test.exs b/test/broadway/producer_test.exs
index d00a328..5f01268 100644
--- a/test/broadway/producer_test.exs
+++ b/test/broadway/producer_test.exs
@@ -1,5 +1,5 @@
 defmodule Broadway.ProducerTest do
-  use ExUnit.Case
+  use ExUnit.Case, async: true
 
   alias Broadway.{Producer, Message}
 
diff --git a/test/broadway/term_storage_test.exs b/test/broadway/term_storage_test.exs
new file mode 100644
index 0000000..27f0c21
--- /dev/null
+++ b/test/broadway/term_storage_test.exs
@@ -0,0 +1,21 @@
+defmodule Broadway.TermStorageTest do
+  use ExUnit.Case, async: true
+
+  doctest Broadway.TermStorage
+  alias Broadway.TermStorage
+
+  test "allows terms to be written and read" do
+    ref = TermStorage.put({:really, :unique, :term})
+    assert TermStorage.get!(ref) == {:really, :unique, :term}
+  end
+
+  test "returns the same reference for the same term" do
+    assert TermStorage.put({:really, :unique, :term}) ==
+             TermStorage.put({:really, :unique, :term})
+  end
+
+  test "returns different references for different terms" do
+    assert TermStorage.put({:really, :unique, :term}) !=
+             TermStorage.put({:another, :unique, :term})
+  end
+end