Address PR feedback

Author: maartenvanvliet

lib/gen_stage.ex

@@ -1121,10 +1121,12 @@ defmodule GenStage do
             when new_state: term, event: term
 
   @doc """
-  Invoked to handle `continue` instructions.
+  Invoked to handle `:continue` instructions.
 
-  It is useful for performing work after initialization or for splitting the work
-  in a callback in multiple steps, updating the process state along the way.
+  This callback can be used to perform work right after emitting events from
+  other callbacks. The "continue mechanism" makes sure that no messages,
+  calls, casts, or anything else will be handled between a callback emitting
+  a `:continue` tuple and the `c:handle_continue/2` callback being invoked.
 
   Return values are the same as `c:handle_cast/2`.
 
@@ -1759,56 +1761,29 @@ defmodule GenStage do
       {:producer, state} ->
         init_producer(mod, [], state, nil)
 
-      {:producer, state, {:continue, _term} = continue} ->
-        init_producer(mod, [], state, continue)
-
-      {:producer, state, :hibernate} ->
-        init_producer(mod, [], state, :hibernate)
-
       {:producer, state, opts} when is_list(opts) ->
         init_producer(mod, opts, state, nil)
 
-      {:producer, state, {:continue, _term} = continue, opts} when is_list(opts) ->
-        init_producer(mod, opts, state, continue)
-
-      {:producer, state, :hibernate, opts} when is_list(opts) ->
-        init_producer(mod, opts, state, :hibernate)
+      {:producer, state, opts, continue_or_hibernate} when is_list(opts) ->
+        init_producer(mod, opts, state, continue_or_hibernate)
 
       {:producer_consumer, state} ->
         init_producer_consumer(mod, [], state, nil)
 
-      {:producer_consumer, state, {:continue, _term} = continue} ->
-        init_producer_consumer(mod, [], state, continue)
-
-      {:producer_consumer, state, :hibernate} ->
-        init_producer_consumer(mod, [], state, :hibernate)
-
       {:producer_consumer, state, opts} when is_list(opts) ->
         init_producer_consumer(mod, opts, state, nil)
 
-      {:producer_consumer, state, {:continue, _term} = continue, opts} when is_list(opts) ->
-        init_producer_consumer(mod, opts, state, continue)
-
-      {:producer_consumer, state, :hibernate, opts} when is_list(opts) ->
-        init_producer_consumer(mod, opts, state, :hibernate)
+      {:producer_consumer, state, opts, continue_or_hibernate} when is_list(opts) ->
+        init_producer_consumer(mod, opts, state, continue_or_hibernate)
 
       {:consumer, state} ->
         init_consumer(mod, [], state, nil)
 
-      {:consumer, state, {:continue, _term} = continue} ->
-        init_consumer(mod, [], state, continue)
-
-      {:consumer, state, :hibernate} ->
-        init_consumer(mod, [], state, :hibernate)
-
       {:consumer, state, opts} when is_list(opts) ->
         init_consumer(mod, opts, state, nil)
 
-      {:consumer, state, {:continue, _term} = continue, opts} when is_list(opts) ->
-        init_consumer(mod, opts, state, continue)
-
-      {:consumer, state, :hibernate, opts} when is_list(opts) ->
-        init_consumer(mod, opts, state, :hibernate)
+      {:consumer, state, opts, continue_or_hibernate} when is_list(opts) ->
+        init_consumer(mod, opts, state, continue_or_hibernate)
 
       {:stop, _} = stop ->
         stop

test/gen_stage_test.exs

@@ -442,7 +442,7 @@ defmodule GenStageTest do
   if otp_version >= 21 do
     describe "handle_continue tests" do
       test "producing_init with continue instruction setting counter start position" do
-        {:ok, producer} = Counter.start_link({:producer, 0, {:continue, 500}})
+        {:ok, producer} = Counter.start_link({:producer, 0, [], {:continue, 500}})
         {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer]})
 
         batch = Enum.to_list(0..499)
@@ -454,7 +454,7 @@ defmodule GenStageTest do
       end
 
       test "producer_init with nested continue instruction setting counter start position" do
-        {:ok, producer} = CounterNestedContinue.start_link({:producer, 0, {:continue, 500}})
+        {:ok, producer} = CounterNestedContinue.start_link({:producer, 0, [], {:continue, 500}})
         {:ok, _} = Forwarder.start_link({:consumer, self(), subscribe_to: [producer]})
 
         # The nested continue sets the counter to 2000
@@ -471,11 +471,11 @@ defmodule GenStageTest do
       end
 
       test "consumer_init with continue instruction" do
-        {:ok, producer} = Counter.start_link({:producer, 0, {:continue, 500}})
+        {:ok, producer} = Counter.start_link({:producer, 0, [], {:continue, 500}})
 
         {:ok, _} =
           Forwarder.start_link(
-            {:consumer, self(), {:continue, :continue_reached}, subscribe_to: [producer]}
+            {:consumer, self(), [subscribe_to: [producer]], {:continue, :continue_reached}}
           )
 
         assert_receive :continue_reached
@@ -486,8 +486,9 @@ defmodule GenStageTest do
 
         {:ok, _doubler} =
           Doubler.start_link(
-            {:producer_consumer, self(), {:continue, :continue_reached},
-             subscribe_to: [{producer, max_demand: 100, min_demand: 80}]}
+            {:producer_consumer, self(),
+             [subscribe_to: [{producer, max_demand: 100, min_demand: 80}]],
+             {:continue, :continue_reached}}
           )
 
         assert_receive :continue_reached
@@ -497,7 +498,7 @@ defmodule GenStageTest do
 
   describe "hibernate tests" do
     test "producer_init with hibernate instruction" do
-      {:ok, producer} = Counter.start_link({:producer, 0, :hibernate})
+      {:ok, producer} = Counter.start_link({:producer, 0, [], :hibernate})
 
       assert :erlang.process_info(producer, :current_function) ==
                {:current_function, {:erlang, :hibernate, 3}}
@@ -507,7 +508,7 @@ defmodule GenStageTest do
       {:ok, producer} = Counter.start_link({:producer, 0})
 
       {:ok, consumer} =
-        Forwarder.start_link({:consumer, self(), :hibernate, subscribe_to: [producer]})
+        Forwarder.start_link({:consumer, self(), [subscribe_to: [producer]], :hibernate})
 
       assert :erlang.process_info(consumer, :current_function) ==
                {:current_function, {:erlang, :hibernate, 3}}
@@ -518,8 +519,8 @@ defmodule GenStageTest do
 
       {:ok, doubler} =
         Doubler.start_link(
-          {:producer_consumer, self(), :hibernate,
-           subscribe_to: [{producer, max_demand: 100, min_demand: 80}]}
+          {:producer_consumer, self(),
+           [subscribe_to: [{producer, max_demand: 100, min_demand: 80}]], :hibernate}
         )
 
       assert :erlang.process_info(doubler, :current_function) ==