Repository: plataformatec/broadway
Pull request #47 (2019-02-18 18:18:41)

Simplify producer config
Instead of:

```elixir
producers: [
  default: [
    module: SQSProducer,
    arg: [
      sqs_client: {ExAwsClient, [
        queue_name: "ptec",
      ]}
    ],
    stages: 1
  ],
],
```


```elixir
producers: [
  default: [
    module: {SQSProducer, queue_name: "ptec"},
    stages: 1
  ],
],
```


josevalim: Ship it!!!!
-- 


*José Valim*
www.plataformatec.com.br
Skype: jv.ptec
Founder and Director of R&D


Patch:
diff --git a/README.md b/README.md
index 4c739fc..dc6bbb6 100644
--- a/README.md
+++ b/README.md
@@ -39,15 +39,13 @@ defmodule MyBroadway do
   use Broadway
 
   alias Broadway.Message
-  alias BroadwaySQS.{SQSProducer, ExAwsClient}
 
   def start_link(_opts) do
     Broadway.start_link(__MODULE__,
       name: __MODULE__,
       producers: [
         sqs: [
-          module: SQSProducer,
-          arg: [sqs_client: {ExAwsClient, [queue_name: "my_queue"]}]
+          module: {BroadwaySQS.Producer, queue_name: "my_queue"}
         ]
       ],
       processors: [
diff --git a/guides/examples/Custom Producers.md b/guides/examples/Custom Producers.md
index 6e9ae6f..acf05cd 100644
--- a/guides/examples/Custom Producers.md
+++ b/guides/examples/Custom Producers.md
@@ -45,8 +45,7 @@ generated by the producer into proper Broadway messages:
           name: __MODULE__,
           producers: [
             default: [
-              module: Counter,
-              arg: 1,
+              module: {Counter, 1},
               transformer: {__MODULE__, :transform, []}
             ],
           ],
@@ -73,15 +72,14 @@ generated by the producer into proper Broadway messages:
       end
     end
 
-Notice that you need to pass three options to the producer:
+Notice that you need to pass two options to the producer:
 
-  * `:module` - the module that implements the GenStage behaviour.
-
-  * `:arg` - the argument that will be given to the `init` callback
-    of the GenStage. It is very important to note that Broadway
-    **will not call** the `child_spec/1` or `start_link/1` function
-    of the producer. That's because Broadway wraps the producer to
-    augment it with extra features.
+  * `:module` - a tuple representing the GenStage producer as `{mod, arg}`.
+    Where `mod` is module that implements the GenStage behaviour and `arg`
+    the argument that will be given to the `init` callback of the GenStage.
+    It is very important to note that Broadway **will not call** the
+    `child_spec/1` or `start_link/1` function of the producer. That's
+    because Broadway wraps the producer to augment it with extra features.
 
   * `:transformer` - a module-function-args tuple that will be invoked
     inside the producer, for every producer message, that should create
diff --git a/lib/broadway.ex b/lib/broadway.ex
index 5c36d08..4b5aa59 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -60,7 +60,10 @@ defmodule Broadway do
           Broadway.start_link(MyBroadway,
             name: MyBroadwayExample,
             producers: [
-              default: [module: Counter, stages: 1]
+              default: [
+                module: {Counter, []},
+                stages: 1
+              ]
             ],
             processors: [
               default: [stages: 2]
@@ -354,11 +357,11 @@ defmodule Broadway do
 
   The producer options are:
 
-    * `:module` - Required. A module defining a GenStage producer.
-      Pay attention that this producer must emit events that are
-      `Broadway.Message` structs.
-    * `:arg` - Required. The argument that will be passed to the
-      `init/1` callback of the producer.
+    * `:module` - Required. A tuple representing a GenStage producer.
+      The tuple format should be `{mod, arg}`, where `mod` is the module
+      that implements the GenStage behaviour and `arg` the argument that will
+      be passed to the `init/1` callback of the producer. Pay attention that
+      this producer must emit events that are `Broadway.Message` structs.
     * `:stages` - Optional. The number of stages that will be
       created by Broadway. Use this option to control the concurrency
       level of each set of producers. The default value is `1`.
@@ -465,8 +468,7 @@ defmodule Broadway do
         type: :keyword_list,
         keys: [
           *: [
-            module: [required: true, type: :atom],
-            arg: [required: true],
+            module: [required: true, type: :mod_arg],
             stages: [type: :pos_integer, default: 1],
             transformer: [type: :mfa, default: nil]
           ]
diff --git a/lib/broadway/options.ex b/lib/broadway/options.ex
index 3f40c57..6ee98ab 100644
--- a/lib/broadway/options.ex
+++ b/lib/broadway/options.ex
@@ -1,7 +1,7 @@
 defmodule Broadway.Options do
   @moduledoc false
 
-  @types [:any, :keyword_list, :atom, :non_neg_integer, :pos_integer, :mfa]
+  @types [:any, :keyword_list, :atom, :non_neg_integer, :pos_integer, :mfa, :mod_arg]
 
   def validate(opts, spec) do
     case validate_unknown_options(opts, spec) do
@@ -105,11 +105,22 @@ defmodule Broadway.Options do
     end
   end
 
-  defp validate_type(:mfa, key, {m, f, args} = value)
-       when not is_atom(m) or not is_atom(f) or not is_list(args) do
+  defp validate_type(:mfa, _key, {m, f, args}) when is_atom(m) and is_atom(f) and is_list(args) do
+    :ok
+  end
+
+  defp validate_type(:mfa, key, value) when not is_nil(value) do
     {:error, "expected #{inspect(key)} to be a tuple {Mod, Fun, Args}, got: #{inspect(value)}"}
   end
 
+  defp validate_type(:mod_arg, _key, {m, _arg}) when is_atom(m) do
+    :ok
+  end
+
+  defp validate_type(:mod_arg, key, value) do
+    {:error, "expected #{inspect(key)} to be a tuple {Mod, Arg}, got: #{inspect(value)}"}
+  end
+
   defp validate_type(nil, key, value) do
     validate_type(:any, key, value)
   end
diff --git a/lib/broadway/producer.ex b/lib/broadway/producer.ex
index 7d1d1dc..f6f2a15 100644
--- a/lib/broadway/producer.ex
+++ b/lib/broadway/producer.ex
@@ -13,10 +13,10 @@ defmodule Broadway.Producer do
 
   @impl true
   def init(args) do
-    module = args[:module]
+    {module, arg} = args[:module]
     transformer = args[:transformer]
     # TODO: Raise a proper error message if we don't get  {:producer, state} back.
-    {:producer, module_state} = module.init(args[:arg])
+    {:producer, module_state} = module.init(arg)
 
     state = %{
       module: module,


Test patch:
diff --git a/test/broadway/options_test.exs b/test/broadway/options_test.exs
index 4e0f0fa..9c563ab 100644
--- a/test/broadway/options_test.exs
+++ b/test/broadway/options_test.exs
@@ -122,6 +122,41 @@ defmodule Broadway.OptionsTest do
                :error,
                ~s(expected :transformer to be a tuple {Mod, Fun, Args}, got: {SomeMod, :func, "not_a_list"})
              }
+
+      opts = [transformer: NotATuple]
+
+      assert Options.validate(opts, spec) == {
+               :error,
+               ~s(expected :transformer to be a tuple {Mod, Fun, Args}, got: NotATuple)
+             }
+    end
+
+    test "valid mod_arg" do
+      spec = [producer: [type: :mod_arg]]
+
+      opts = [producer: {SomeMod, [1, 2]}]
+      assert Options.validate(opts, spec) == {:ok, opts}
+
+      opts = [producer: {SomeMod, []}]
+      assert Options.validate(opts, spec) == {:ok, opts}
+    end
+
+    test "invalid mod_arg" do
+      spec = [producer: [type: :mod_arg]]
+
+      opts = [producer: NotATuple]
+
+      assert Options.validate(opts, spec) == {
+               :error,
+               ~s(expected :producer to be a tuple {Mod, Arg}, got: NotATuple)
+             }
+
+      opts = [producer: {"not_a_module", []}]
+
+      assert Options.validate(opts, spec) == {
+               :error,
+               ~s(expected :producer to be a tuple {Mod, Arg}, got: {"not_a_module", []})
+             }
     end
   end
 
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index 2305b0f..06db57e 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -142,7 +142,7 @@ defmodule BroadwayTest do
         name: broadway,
         context: %{test_pid: self()},
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [default: []]
@@ -158,7 +158,7 @@ defmodule BroadwayTest do
         name: broadway,
         context: %{test_pid: self()},
         producers: [
-          default: [module: ManualProducer, arg: [], stages: 3]
+          default: [module: {ManualProducer, []}, stages: 3]
         ],
         processors: [default: []],
         batchers: [default: []]
@@ -173,7 +173,7 @@ defmodule BroadwayTest do
       Broadway.start_link(Forwarder,
         name: broadway,
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [default: []]
@@ -188,7 +188,7 @@ defmodule BroadwayTest do
       Broadway.start_link(Forwarder,
         name: broadway,
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: [stages: 13]],
         batchers: [default: []]
@@ -204,7 +204,7 @@ defmodule BroadwayTest do
         name: broadway,
         context: %{test_pid: self()},
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [p1: [], p2: []]
@@ -221,7 +221,7 @@ defmodule BroadwayTest do
         name: broadway,
         context: %{test_pid: self()},
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [
@@ -240,7 +240,7 @@ defmodule BroadwayTest do
       Broadway.start_link(Forwarder,
         name: broadway,
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [default: []]
@@ -257,7 +257,7 @@ defmodule BroadwayTest do
         name: new_unique_name(),
         context: %{test_pid: self()},
         producers: [
-          default: [module: ManualProducer, arg: []]
+          default: [module: {ManualProducer, []}]
         ],
         processors: [default: []],
         batchers: [default: []]
@@ -303,7 +303,7 @@ defmodule BroadwayTest do
           name: broadway_name,
           context: context,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 2]],
           batchers: [default: [batch_size: 2]]
@@ -418,7 +418,7 @@ defmodule BroadwayTest do
           name: broadway_name,
           context: context,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: []],
           batchers: [
@@ -473,7 +473,7 @@ defmodule BroadwayTest do
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           context: context,
-          producers: [default: [module: ManualProducer, arg: []]],
+          producers: [default: [module: {ManualProducer, []}]],
           processors: [default: []],
           batchers: [default: [batch_size: 2, batch_timeout: 20]]
         )
@@ -510,8 +510,7 @@ defmodule BroadwayTest do
           context: %{test_pid: self()},
           producers: [
             default: [
-              module: EventProducer,
-              arg: Map.get(tags, :events, [1, 2, 3]),
+              module: {EventProducer, Map.get(tags, :events, [1, 2, 3])},
               transformer: {Transformer, :transform, test_pid: self()}
             ]
           ],
@@ -566,8 +565,7 @@ defmodule BroadwayTest do
           context: context,
           producers: [
             default: [
-              module: ManualProducer,
-              arg: %{test_pid: self()}
+              module: {ManualProducer, %{test_pid: self()}}
             ]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 2]],
@@ -648,7 +646,7 @@ defmodule BroadwayTest do
           name: broadway_name,
           context: context,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 2]],
           batchers: [default: [batch_size: 2]]
@@ -755,7 +753,7 @@ defmodule BroadwayTest do
           name: broadway_name,
           context: context,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 2]],
           batchers: [default: [batch_size: 2]]
@@ -858,7 +856,7 @@ defmodule BroadwayTest do
           name: broadway_name,
           context: context,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 4]],
           batchers: [default: [batch_size: 4]]
@@ -953,7 +951,7 @@ defmodule BroadwayTest do
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           producers: [
-            default: [module: ManualProducer, arg: []]
+            default: [module: {ManualProducer, []}]
           ],
           processors: [default: [stages: 1, min_demand: 1, max_demand: 4]],
           batchers: [default: [batch_size: 4]],