Missing handle info (#245)

Log any extra messages received and ignore them. Fixes #238.

Author: mgwidmann

lib/gen_stage.ex

@@ -1605,9 +1605,10 @@ defmodule GenStage do
   """
   @spec from_enumerable(Enumerable.t(), keyword) :: GenServer.on_start()
   def from_enumerable(stream, opts \\ []) do
+    {:current_stacktrace, [_info_call | stack]} = Process.info(self(), :current_stacktrace)
     case Keyword.pop(opts, :link, true) do
-      {true, opts} -> start_link(GenStage.Streamer, {stream, opts}, opts)
-      {false, opts} -> start(GenStage.Streamer, {stream, opts}, opts)
+      {true, opts} -> start_link(GenStage.Streamer, {{stream, opts}, stack}, opts)
+      {false, opts} -> start(GenStage.Streamer, {{stream, opts}, stack}, opts)
     end
   end
 

lib/gen_stage/streamer.ex

@@ -3,35 +3,42 @@ defmodule GenStage.Streamer do
   use GenStage
 
   def start_link({_, opts} = pair) do
-    GenStage.start_link(__MODULE__, pair, opts)
+    {:current_stacktrace, [_info_call | stack]} = Process.info(self(), :current_stacktrace)
+    GenStage.start_link(__MODULE__, {pair, stack}, opts)
   end
 
-  def init({stream, opts}) do
+  def init({{stream, opts}, stack}) do
     continuation =
       &Enumerable.reduce(stream, &1, fn
         x, {acc, 1} -> {:suspend, {[x | acc], 0}}
         x, {acc, counter} -> {:cont, {[x | acc], counter - 1}}
       end)
 
-    {:producer, continuation, Keyword.take(opts, [:dispatcher, :demand])}
+    {:producer, {stack, continuation}, Keyword.take(opts, [:dispatcher, :demand])}
   end
 
-  def handle_demand(_demand, continuation) when is_atom(continuation) do
-    {:noreply, [], continuation}
+  def handle_demand(_demand, {stack, continuation}) when is_atom(continuation) do
+    {:noreply, [], {stack, continuation}}
   end
 
-  def handle_demand(demand, continuation) when demand > 0 do
+  def handle_demand(demand, {stack, continuation}) when demand > 0 do
     case continuation.({:cont, {[], demand}}) do
       {:suspended, {list, 0}, continuation} ->
-        {:noreply, :lists.reverse(list), continuation}
+        {:noreply, :lists.reverse(list), {stack, continuation}}
 
       {status, {list, _}} ->
         GenStage.async_info(self(), :stop)
-        {:noreply, :lists.reverse(list), status}
+        {:noreply, :lists.reverse(list), {stack, status}}
     end
   end
 
   def handle_info(:stop, state) do
     {:stop, :normal, state}
   end
+
+  def handle_info(msg, {stack, continuation}) do
+    log = '** Undefined handle_info in ~tp~n** Unhandled message: ~tp~n** Stream started at:~n~ts'
+    :error_logger.warning_msg(log, [inspect(__MODULE__), msg, Exception.format_stacktrace(stack)])
+    {:noreply, [], {stack, continuation}}
+  end
 end