Repository: plataformatec/broadway
Pull request #22 (2019-01-30 00:09:34)

Pass context as option instead of argument

josevalim: @msaraiva can you please describe the rationale for the change? I don't remember anymore. :D We should probably have it as part of the merge message. But code wise it looks good. :+1:
msaraiva: @josevalim the `context` exists mostly for tests (at least it was the initial idea), so we discussed a while ago that we shouldn't force the user to always pass it as a required argument.
josevalim: I don't think it will be set only for testing but I don't guess yours and mine opinion matters here because the best we can do is guess.

If we try to be a bit more objective, I would say that a positive thing about it being an explicit argument is that it mirrors GenServer, Supervisor and other behaviours... although it may be a false mirrorring since our context is immutable.

Can you think of other pros and cons?
msaraiva: If keeping the context as an argument gives the user the feeling that he's defining an initial mutable state (just like GenServers and friends), then I think this is probably the best reason to move it to the options. I agree that we don't know yet how often the context is going to be used, but still, not having as an argument sounds to me as the best initial approach.
josevalim: So maybe all we need is a better default? Maybe we should set it to `:context_was_not_set` or something? To clarify, my concern is that this value is always given to the user code (i.e. it is required) but we are treating it as something optional. So there is a mismatch here. The user will be most likely asking "from where this did value come from?".

msaraiva: I like it! Would `:context_not_set` be enough?
josevalim: :+1:

Patch:
diff --git a/README.md b/README.md
index 40f5728..03471b5 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@ end
     alias BroadwaySQS.{SQSProducer, ExAwsClient}
 
     def start_link(_opts) do
-      Broadway.start_link(__MODULE__, %{},
+      Broadway.start_link(__MODULE__,
         name: __MODULE__,
         producers: [
           sqs: [
diff --git a/lib/broadway.ex b/lib/broadway.ex
index 95bb534..3cd75ba 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -34,7 +34,7 @@ defmodule Broadway do
         use Broadway
 
         def start_link(_opts) do
-          Broadway.start_link(MyBroadway, %{},
+          Broadway.start_link(MyBroadway,
             name: MyBroadwayExample,
             producers: [
               default: [module: Counter, stages: 1]
@@ -292,8 +292,6 @@ defmodule Broadway do
   Starts a `Broadway` process linked to the current process.
 
     * `module` is the module implementing the `Broadway` behaviour.
-    * `context` is an immutable user defined data structure that will
-      be passed to `handle_message/2` and `handle_batch/4`.
 
   ## Options
 
@@ -318,6 +316,9 @@ defmodule Broadway do
       where the key is an atom as identifier and the value is another
       keyword list of options. See "Consumers options" section below.
 
+    * `context` is an immutable user defined data structure that will
+      be passed to `handle_message/2` and `handle_batch/4`.
+
   ### Producers options
 
   The producer options are:
@@ -362,19 +363,20 @@ defmodule Broadway do
       (1 second).
 
   """
-  def start_link(module, context, opts) do
+  def start_link(module, opts) do
     case Options.validate(opts, configuration_spec()) do
       {:error, message} ->
         raise ArgumentError, "invalid configuration given to Broadway.start_link/3, " <> message
 
       {:ok, opts} ->
-        Server.start_link(module, context, opts)
+        Server.start_link(module, opts)
     end
   end
 
   defp configuration_spec() do
     [
       name: [required: true, type: :atom],
+      context: [type: :any, default: :context_not_set],
       producers: [
         required: true,
         type: :keyword_list,
diff --git a/lib/broadway/server.ex b/lib/broadway/server.ex
index 95041f7..a1149ac 100644
--- a/lib/broadway/server.ex
+++ b/lib/broadway/server.ex
@@ -4,14 +4,14 @@ defmodule Broadway.Server do
 
   alias Broadway.{Producer, Processor, Batcher, Consumer}
 
-  def start_link(module, context, opts) do
-    GenServer.start_link(__MODULE__, {module, context, opts}, opts)
+  def start_link(module, opts) do
+    GenServer.start_link(__MODULE__, {module, opts}, opts)
   end
 
   @impl true
-  def init({module, context, opts}) do
+  def init({module, opts}) do
     Process.flag(:trap_exit, true)
-    {:ok, supervisor_pid} = start_supervisor(module, context, opts)
+    {:ok, supervisor_pid} = start_supervisor(module, opts)
     {:ok, %{supervisor_pid: supervisor_pid}}
   end
 
@@ -35,8 +35,8 @@ defmodule Broadway.Server do
     end
   end
 
-  defp start_supervisor(module, context, opts) do
-    %{name: broadway_name} = config = init_config(module, context, opts)
+  defp start_supervisor(module, opts) do
+    %{name: broadway_name} = config = init_config(module, opts)
 
     supervisor_name = Module.concat(broadway_name, "Supervisor")
     {producers_names, producers_specs} = build_producers_specs(config)
@@ -54,14 +54,14 @@ defmodule Broadway.Server do
     Supervisor.start_link(children, name: supervisor_name, strategy: :one_for_one)
   end
 
-  defp init_config(module, context, opts) do
+  defp init_config(module, opts) do
     %{
       name: opts[:name],
       module: module,
       processors_config: opts[:processors],
       producers_config: opts[:producers],
       publishers_config: opts[:publishers],
-      context: context
+      context: opts[:context]
     }
   end
 


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index e44b562..7833789 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -108,7 +108,7 @@ defmodule BroadwayTest do
         ArgumentError,
         "invalid configuration given to Broadway.start_link/3, expected :name to be an atom, got: 1",
         fn ->
-          Broadway.start_link(Forwarder, %{}, name: 1)
+          Broadway.start_link(Forwarder, name: 1)
         end
       )
     end
@@ -116,8 +116,9 @@ defmodule BroadwayTest do
     test "default number of producers is 1" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{test_pid: self()},
+      Broadway.start_link(Forwarder,
         name: broadway,
+        context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: []]
         ],
@@ -131,8 +132,9 @@ defmodule BroadwayTest do
     test "set number of producers" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{test_pid: self()},
+      Broadway.start_link(Forwarder,
         name: broadway,
+        context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: [], stages: 3]
         ],
@@ -146,7 +148,7 @@ defmodule BroadwayTest do
     test "default number of processors is schedulers_online * 2" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{},
+      Broadway.start_link(Forwarder,
         name: broadway,
         producers: [
           default: [module: ManualProducer, arg: []]
@@ -161,7 +163,7 @@ defmodule BroadwayTest do
     test "set number of processors" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{},
+      Broadway.start_link(Forwarder,
         name: broadway,
         producers: [
           default: [module: ManualProducer, arg: []]
@@ -176,8 +178,9 @@ defmodule BroadwayTest do
     test "default number of consumers is 1 per publisher" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{test_pid: self()},
+      Broadway.start_link(Forwarder,
         name: broadway,
+        context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: []]
         ],
@@ -192,8 +195,9 @@ defmodule BroadwayTest do
     test "set number of consumers" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{test_pid: self()},
+      Broadway.start_link(Forwarder,
         name: broadway,
+        context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: []]
         ],
@@ -207,14 +211,31 @@ defmodule BroadwayTest do
       assert get_n_consumers(broadway, :p1) == 2
       assert get_n_consumers(broadway, :p2) == 3
     end
+
+    test "default context is :context_not_set" do
+      broadway = new_unique_name()
+
+      Broadway.start_link(Forwarder,
+        name: broadway,
+        producers: [
+          default: [module: ManualProducer, arg: []]
+        ],
+        processors: [],
+        publishers: [default: []]
+      )
+
+      pid = get_processor(broadway, 1)
+      assert :sys.get_state(pid).state.context == :context_not_set
+    end
   end
 
   describe "producer" do
     test "push_messages/2" do
       broadway = new_unique_name()
 
-      Broadway.start_link(Forwarder, %{test_pid: self()},
+      Broadway.start_link(Forwarder,
         name: broadway,
+        context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: []]
         ],
@@ -242,8 +263,9 @@ defmodule BroadwayTest do
       broadway = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(Forwarder, %{test_pid: self()},
+        Broadway.start_link(Forwarder,
           name: broadway,
+          context: %{test_pid: self()},
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -305,8 +327,9 @@ defmodule BroadwayTest do
       broadway_name = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(ForwarderWithCustomHandlers, context,
+        Broadway.start_link(ForwarderWithCustomHandlers,
           name: broadway_name,
+          context: context,
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -358,8 +381,9 @@ defmodule BroadwayTest do
       broadway = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(Forwarder, %{test_pid: self()},
+        Broadway.start_link(Forwarder,
           name: broadway,
+          context: %{test_pid: self()},
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -443,8 +467,9 @@ defmodule BroadwayTest do
       broadway_name = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(ForwarderWithCustomHandlers, context,
+        Broadway.start_link(ForwarderWithCustomHandlers,
           name: broadway_name,
+          context: context,
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -534,8 +559,9 @@ defmodule BroadwayTest do
       broadway_name = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(ForwarderWithCustomHandlers, context,
+        Broadway.start_link(ForwarderWithCustomHandlers,
           name: broadway_name,
+          context: context,
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -610,8 +636,9 @@ defmodule BroadwayTest do
       broadway_name = new_unique_name()
 
       {:ok, _pid} =
-        Broadway.start_link(ForwarderWithCustomHandlers, context,
+        Broadway.start_link(ForwarderWithCustomHandlers,
           name: broadway_name,
+          context: context,
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -652,8 +679,9 @@ defmodule BroadwayTest do
       Process.flag(:trap_exit, true)
 
       {:ok, pid} =
-        Broadway.start_link(Forwarder, %{test_pid: self()},
+        Broadway.start_link(Forwarder,
           name: new_unique_name(),
+          context: %{test_pid: self()},
           producers: [
             default: [module: ManualProducer, arg: []]
           ],
@@ -674,8 +702,9 @@ defmodule BroadwayTest do
       Process.flag(:trap_exit, true)
 
       {:ok, pid} =
-        Broadway.start_link(Forwarder, %{test_pid: self()},
+        Broadway.start_link(Forwarder,
           name: new_unique_name(),
+          context: %{test_pid: self()},
           producers: [
             default: [module: ManualProducer, arg: []]
           ],