Make :subscribe_to option type less restrictive (#261)

Author: bamorim

lib/gen_stage.ex

@@ -792,7 +792,7 @@ defmodule GenStage do
 
   @typedoc "Option values used by the `init*` common to `:consumer` and `:producer_consumer` types"
   @type consumer_and_producer_consumer_option ::
-          {:subscribe_to, [module | {module, subscription_options}]}
+          {:subscribe_to, [atom | pid | {GenServer.server(), subscription_options}]}
 
   @typedoc "Option values used by the `init*` functions when stage type is `:producer`"
   @type producer_option :: producer_only_option | producer_and_producer_consumer_option
@@ -2419,6 +2419,8 @@ defmodule GenStage do
   defp consumer_init_subscribe(producers, stage) do
     fold_fun = fn
       to, {:ok, stage} ->
+        maybe_print_subscribe_to_deprecation_warning(to)
+
         case consumer_subscribe(to, stage) do
           {:reply, _, stage} -> {:ok, stage}
           {:stop, reason, _, _} -> {:stop, reason}
@@ -2432,6 +2434,21 @@ defmodule GenStage do
     :lists.foldl(fold_fun, {:ok, stage}, producers)
   end
 
+  defp maybe_print_subscribe_to_deprecation_warning(to) do
+    log = ':subscribe_to value with type ~ts is deprecated. Change ~tp to {~tp, []} instead.'
+
+    case to do
+      {:global, _} ->
+        :error_logger.warning_msg(log, ['{:global, term()}', to, to])
+
+      {:via, _, _} ->
+        :error_logger.warning_msg(log, ['{:via, module(), term()}', to, to])
+
+      _ ->
+        :ok
+    end
+  end
+
   defp consumer_receive({_, ref} = from, {producer_id, cancel, {demand, min, max}}, events, stage) do
     {demand, batches} = Utils.split_batches(events, from, min, max, demand)
     stage = put_in(stage.producers[ref], {producer_id, cancel, {demand, min, max}})

test/gen_stage_test.exs

@@ -1868,4 +1868,51 @@ defmodule GenStageTest do
                {:registered_name, :gen_stage_from_enumerable}
     end
   end
+
+  describe "subscribe_to names" do
+    test "can be pid" do
+      {:ok, producer} = Counter.start_link({:producer, 0})
+      {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer]})
+      assert_receive {:consumed, _}
+    end
+
+    test "can be an atom name" do
+      producer_name = :producer_atom_name
+      {:ok, _} = Counter.start_link({:producer, 0}, name: producer_name)
+      {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer_name]})
+      assert_receive {:consumed, _}
+    end
+
+    test "can be a via name if passed with options to avoid ambiguity" do
+      producer_name = {:via, :global, {:producer, :name}}
+      {:ok, _} = Counter.start_link({:producer, 0}, name: producer_name)
+      {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [{producer_name, []}]})
+      assert_receive {:consumed, _}
+    end
+
+    test "can be a global name if passed with options to avoid ambiguity" do
+      producer_name = {:global, {:producer, :name}}
+      {:ok, _} = Counter.start_link({:producer, 0}, name: producer_name)
+      {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [{producer_name, []}]})
+      assert_receive {:consumed, _}
+    end
+
+    test "logs warning about ambiguity if global name is passed without options" do
+      producer_name = {:global, {:producer, :name}}
+      {:ok, _} = Counter.start_link({:producer, 0}, name: producer_name)
+
+      assert capture_log(fn ->
+               {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer_name]})
+             end) =~ ":subscribe_to value with type {:global, term()} is deprecated."
+    end
+
+    test "logs warning about ambiguity if via name is passed without options" do
+      producer_name = {:via, :global, {:producer, :name}}
+      {:ok, _} = Counter.start_link({:producer, 0}, name: producer_name)
+
+      assert capture_log(fn ->
+               {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer_name]})
+             end) =~ ":subscribe_to value with type {:via, module(), term()} is deprecated."
+    end
+  end
 end