Allow only one process handler per process

Author: José Valim

lib/elixir/lib/gen_event.ex

@@ -732,57 +732,33 @@ defmodule GenEvent do
 
   defp server_add_handler({module, id}, args, handlers) do
     handler = handler(module: module, id: {module, id})
-    server_add_handler(module, handler, args, handlers)
+    do_add_handler(module, handler, args, handlers, :ok)
   end
 
   defp server_add_handler(module, args, handlers) do
     handler = handler(module: module, id: module)
-    server_add_handler(module, handler, args, handlers)
-  end
-
-  defp server_add_handler(module, handler, arg, handlers) do
-    case :lists.keyfind(handler(handler, :id), handler(:id) + 1, handlers) do
-      false ->
-        case do_handler(module, :init, [arg]) do
-          {:ok, res} ->
-            case res do
-              {:ok, state} ->
-                {false, :ok, [handler(handler, state: state)|handlers]}
-              {:ok, state, :hibernate} ->
-                {true, :ok, [handler(handler, state: state)|handlers]}
-              {:error, _} = error ->
-                {false, error, handlers}
-              other ->
-                {false, {:error, {:bad_return_value, other}}, handlers}
-            end
-          {:error, _} = error ->
-            {false, error, handlers}
-        end
-      _ ->
-        {false, {:error, :already_added}, handlers}
-    end
+    do_add_handler(module, handler, args, handlers, :ok)
   end
 
   defp server_add_mon_handler({module, id}, args, handlers, pid) do
     ref = Process.monitor(pid)
     handler = handler(module: module, id: {module, id}, pid: pid, ref: ref)
-    server_add_handler(module, handler, args, handlers)
+    do_add_handler(module, handler, args, handlers, :ok)
   end
 
   defp server_add_mon_handler(module, args, handlers, pid) do
     ref = Process.monitor(pid)
     handler = handler(module: module, id: module, pid: pid, ref: ref)
-    server_add_handler(module, handler, args, handlers)
+    do_add_handler(module, handler, args, handlers, :ok)
   end
 
   defp server_add_process_handler(pid, notify, handlers) do
     ref = Process.monitor(pid)
-    {:ok, state} = GenEvent.Stream.init({pid, ref})
     # Notice the pid is set only when notifications
     # are explicitly required.
-    handler = handler(module: GenEvent.Stream, id: {GenEvent.Stream, ref},
-                      pid: if(notify, do: pid), ref: ref, state: state)
-    {false, ref, [handler|handlers]}
+    handler = handler(module: GenEvent.Stream, id: {GenEvent.Stream, pid},
+                      pid: if(notify, do: pid), ref: ref)
+    do_add_handler(GenEvent.Stream, handler, {pid, ref}, handlers, {self(), ref})
   end
 
   defp server_remove_handler(module, args, handlers, name) do
@@ -813,7 +789,7 @@ defmodule GenEvent do
 
   defp server_split_process_handlers(mode, event, [handler|t], handlers, streams) do
     case handler(handler, :id) do
-      {GenEvent.Stream, _ref} ->
+      {GenEvent.Stream, _pid} ->
         server_process_notify(mode, event, handler)
         server_split_process_handlers(mode, event, t, handlers, [handler|streams])
       _ ->
@@ -965,6 +941,29 @@ defmodule GenEvent do
     end
   end
 
+  defp do_add_handler(module, handler, arg, handlers, succ) do
+    case :lists.keyfind(handler(handler, :id), handler(:id) + 1, handlers) do
+      false ->
+        case do_handler(module, :init, [arg]) do
+          {:ok, res} ->
+            case res do
+              {:ok, state} ->
+                {false, succ, [handler(handler, state: state)|handlers]}
+              {:ok, state, :hibernate} ->
+                {true, succ, [handler(handler, state: state)|handlers]}
+              {:error, _} = error ->
+                {false, error, handlers}
+              other ->
+                {false, {:error, {:bad_return_value, other}}, handlers}
+            end
+          {:error, _} = error ->
+            {false, error, handlers}
+        end
+      _ ->
+        {false, {:error, :already_added}, handlers}
+    end
+  end
+
   defp do_swap(handler, args1, module2, args2, name, handlers) do
     pid   = handler(handler, :pid)
     state = do_terminate(handler,

lib/elixir/lib/gen_event/stream.ex

@@ -79,44 +79,30 @@ defimpl Enumerable, for: GenEvent.Stream do
   end
 
   defp start(%{manager: manager} = stream) do
-    {pid, ref, mon_ref} =
-      try do
-        pid = whereis(manager)
-        ref = Process.monitor(pid)
-        {:ok, msg_ref} = :gen.call(pid, self(), {:add_process_handler, self(), true}, :infinity)
-        {pid, msg_ref, ref}
-      catch
-        :exit, reason -> exit({reason, {__MODULE__, :start, [stream]}})
-      end
-
-    {pid, ref, mon_ref}
-  end
-
-  defp whereis(pid) when is_pid(pid), do: pid
-  defp whereis(atom) when is_atom(atom), do: Process.whereis(atom) || exit(:noproc)
-  defp whereis({:global, name}), do: :global.whereis_name(name)
-  defp whereis({:via, module, name}), do: module.whereis_name(name)
-  defp whereis({atom, node}) do
-    case :rpc.call(node, :erlang, :whereis, [atom]) do
-      :undefined -> exit(:noproc)
-      {:badrpc, :nodedown} -> exit({:nodedown, node})
-      pid when is_pid(pid) -> pid
+    try do
+      {:ok, {pid, ref}} = :gen.call(manager, self(), {:add_process_handler, self(), true}, :infinity)
+      mon_ref = Process.monitor(pid)
+      {pid, ref, mon_ref}
+    catch
+      :exit, reason -> exit({reason, {__MODULE__, :start, [stream]}})
     end
   end
 
   defp next(%{timeout: timeout} = stream, {pid, ref, mon_ref} = acc) do
+    self = self()
+
     receive do
       # The handler was removed. Stop iteration, resolve the
       # event later. We need to demonitor now, otherwise DOWN
       # appears with higher priority in the shutdown process.
-      {:gen_event_EXIT, {GenEvent.Stream, ^ref}, _reason} = event ->
+      {:gen_event_EXIT, {GenEvent.Stream, ^self}, _reason} = event ->
         Process.demonitor(mon_ref, [:flush])
-        send(self(), event)
+        send(self, event)
         {:halt, {:removed, acc}}
 
       # The manager died. Stop iteration, resolve the event later.
       {:DOWN, ^mon_ref, _, _, _} = event ->
-        send(self(), event)
+        send(self, event)
         {:halt, {:removed, acc}}
 
       # Got an async event.
@@ -149,23 +135,26 @@ defimpl Enumerable, for: GenEvent.Stream do
 
   # If we reach this branch, the handler was not removed yet,
   # so we trigger a request for doing so.
-  defp stop(stream, {pid, ref, _} = acc) do
-    _ = Task.start(fn -> GenEvent.remove_handler(pid, {GenEvent.Stream, ref}, :shutdown) end)
+  defp stop(stream, {pid, _, _} = acc) do
+    parent = self()
+    _ = Task.start(fn -> GenEvent.remove_handler(pid, {GenEvent.Stream, parent}, :shutdown) end)
     stop(stream, {:removed, acc})
   end
 
   defp wait_for_handler_removal(pid, ref, mon_ref) do
+    self = self()
+
     receive do
       {_from, {^pid, ^ref}, {notify, _event}} when notify in [:ack_notify, :sync_notify] ->
         send pid, {ref, :done}
         wait_for_handler_removal(pid, ref, mon_ref)
-      {:gen_event_EXIT, {GenEvent.Stream, ^ref}, reason}
+      {:gen_event_EXIT, {GenEvent.Stream, ^self}, reason}
           when reason == :normal
           when reason == :shutdown
           when tuple_size(reason) == 3 and elem(reason, 0) == :swapped  ->
         Process.demonitor(mon_ref, [:flush])
         :ok
-      {:gen_event_EXIT, {GenEvent.Stream, ^ref}, reason} ->
+      {:gen_event_EXIT, {GenEvent.Stream, ^self}, reason} ->
         Process.demonitor(mon_ref, [:flush])
         {:error, reason}
       {:DOWN, ^mon_ref, _, _, reason} ->