pulse_libs

defmodule EQC.Pulse.GenEvent do
  @compile {:parse_transform, :pulse_instrument}
  @compile {:pulse_replace_module, [{:gen, :pulse_gen}, {:proc_lib, :pulse_proc_lib}]}
  @moduledoc """
  A behaviour module for implementing event handling functionality.

  The event handling model consists of a generic event manager
  process with an arbitrary number of event handlers which are
  added and deleted dynamically.

  An event manager implemented using this module will have a standard
  set of interface functions and include functionality for tracing and
  error reporting. It will also fit into a supervision tree.

  ## Example

  There are many use cases for event handlers. For example, a logging
  system can be built using event handlers where each log message is
  an event and different event handlers can be plugged to handle the
  log messages. One handler may print error messages on the terminal,
  another can write it to a file, while a third one can keep the
  messages in memory (like a buffer) until they are read.

  As an example, let's have a GenEvent that accumulates messages until
  they are collected by an explicit call.

      defmodule LoggerHandler do
        use GenEvent

        # Callbacks

        def handle_event({:log, x}, messages) do
          {:ok, [x|messages]}
        end

        def handle_call(:messages, messages) do
          {:ok, Enum.reverse(messages), []}
        end
      end

      {:ok, pid} = GenEvent.start_link()

      GenEvent.add_handler(pid, LoggerHandler, [])
      #=> :ok

      GenEvent.notify(pid, {:log, 1})
      #=> :ok

      GenEvent.notify(pid, {:log, 2})
      #=> :ok

      GenEvent.call(pid, LoggerHandler, :messages)
      #=> [1, 2]

      GenEvent.call(pid, LoggerHandler, :messages)
      #=> []

  We start a new event manager by calling `GenEvent.start_link/0`.
  Notifications can be sent to the event manager which will then
  invoke `handle_event/2` for each registered handler.

  We can add new handlers with `add_handler/3` and `add_mon_handler/3`.
  Calls can also be made to specific handlers by using `call/3`.

  ## Callbacks

  There are 6 callbacks required to be implemented in a `GenEvent`. By
  adding `use GenEvent` to your module, Elixir will automatically define
  all 6 callbacks for you, leaving it up to you to implement the ones
  you want to customize. The callbacks are:

    * `init(args)` - invoked when the event handler is added.

      It must return:

      -  `{:ok, state}`
      -  `{:ok, state, :hibernate}`
      -  `{:error, reason}`

    * `handle_event(msg, state)` - invoked whenever an event is sent via
      `notify/2`, `ack_notify/2` or `sync_notify/2`.

      It must return:

      -  `{:ok, new_state}`
      -  `{:ok, new_state, :hibernate}`
      -  `:remove_handler`

    * `handle_call(msg, state)` - invoked when a `call/3` is done to a specific
      handler.

      It must return:

      -  `{:ok, reply, new_state}`
      -  `{:ok, reply, new_state, :hibernate}`
      -  `{:remove_handler, reply}`

    * `handle_info(msg, state)` - invoked to handle all other messages which
      are received by the process. Must return the same values as
      `handle_event/2`.

    * `terminate(reason, state)` - called when the event handler is removed or
      the event manager is terminating. It can return any term.

      The reason is one of:

      -  `:stop` - manager is terminating
      -  `{:stop, reason}` - monitored process terminated (for monitored handlers)
      -  `:remove_handler` - handler is being removed
      -  `{:error, term}` - handler crashed or returned a bad value
      -  `term` - any term passed to functions like `GenEvent.remove_handler/2`

    * `code_change(old_vsn, state, extra)` - called when the application
      code is being upgraded live (hot code swapping).

      It must return:

      -  `{:ok, new_state}`

  ## Name Registration

  A GenEvent is bound to the same name registration rules as a `GenServer`.
  Read more about it in the `GenServer` docs.

  ## Modes

  GenEvent stream supports three different notifications.

  On `GenEvent.ack_notify/2`, the manager acknowledges each event,
  providing back pressure, but processing of the message happens
  asynchronously.

  On `GenEvent.sync_notify/2`, the manager acknowledges an event
  just after it was processed by all event handlers.

  On `GenEvent.notify/2`, all events are processed asynchronously and
  there is no ack (which means there is no backpressure).

  ## Streaming

  `GenEvent` messages can be streamed with the help of `stream/2`.
  Here are some examples:

      stream = GenEvent.stream(pid)

      # Discard the next 10 events
      _ = Enum.drop(stream, 10)

      # Print all remaining events
      for event <- stream do
        IO.inspect event
      end

  ## Learn more and compatibility

  If you wish to find out more about gen events, Elixir getting started
  guides provide a tutorial-like introduction. The documentation and links
  in Erlang can also provide extra insight.

    * http://elixir-lang.org/getting_started/mix/1.html
    * http://www.erlang.org/doc/man/gen_event.html
    * http://learnyousomeerlang.com/event-handlers

  Keep in mind though Elixir and Erlang gen events are not 100% compatible.
  The `:gen_event.add_sup_handler/3` is not supported by Elixir's GenEvent,
  which in turn supports `GenEvent.add_mon_handler/3`.

  The benefits of the monitoring approach are described in the "Don't drink
  too much kool aid" section of the "Learn you some Erlang" link above. Due
  to those changes, Elixir's GenEvent does not trap exits by default.

  Futhermore, Elixir's also normalizes the `{:error, _}` tuples returned
  by many functions, in order to be more consistent with themselves and
  the `GenServer` module.
  """

  @typedoc "Return values of `start*` functions"
  @type on_start :: {:ok, pid} | {:error, {:already_started, pid}}

  @typedoc "The GenEvent manager name"
  @type name :: atom | {:global, term} | {:via, module, term}

  @typedoc "Options used by the `start*` functions"
  @type options :: [name: name]

  @typedoc "The event manager reference"
  @type manager :: pid | name | {atom, node}

  @typedoc "Supported values for new handlers"
  @type handler :: atom | {atom, term} | {pid, reference}

  @doc false
  defmacro __using__(_) do
    quote location: :keep do
      @behaviour :gen_event

      @doc false
      def init(args) do
        {:ok, args}
      end

      @doc false
      def handle_event(_event, state) do
        {:ok, state}
      end

      @doc false
      def handle_call(msg, state) do
        # We do this to trick dialyzer to not complain about non-local returns.
        case :random.uniform(1) do
          1 -> exit({:bad_call, msg})
          2 -> {:remove_handler, :ok}
        end
      end

      @doc false
      def handle_info(_msg, state) do
        {:ok, state}
      end

      @doc false
      def terminate(_reason, _state) do
        :ok
      end

      @doc false
      def code_change(_old, state, _extra) do
        {:ok, state}
      end

      defoverridable [init: 1, handle_event: 2, handle_call: 2,
                      handle_info: 2, terminate: 2, code_change: 3]
    end
  end

  @doc """
  Starts an event manager linked to the current process.

  This is often used to start the `GenEvent` as part of a supervision tree.

  It accepts the `:name` option which is described under the `Name Registration`
  section in the `GenServer` module docs.

  If the event manager is successfully created and initialized, the function
  returns `{:ok, pid}`, where pid is the pid of the server. If there already
  exists a process with the specified server name, the function returns
  `{:error, {:already_started, pid}}` with the pid of that process.

  Note that a `GenEvent` started with `start_link/1` is linked to the
  parent process and will exit not only on crashes but also if the parent
  process exits with `:normal` reason.
  """
  @spec start_link(options) :: on_start
  def start_link(options \\ []) when is_list(options) do
    do_start(:link, options)
  end

  @doc """
  Starts an event manager process without links (outside of a supervision tree).

  See `start_link/1` for more information.
  """
  @spec start(options) :: on_start
  def start(options \\ []) when is_list(options) do
    do_start(:nolink, options)
  end

  @no_callback :"no callback module"

  defp do_start(mode, options) do
    case Keyword.get(options, :name) do
      nil ->
        :gen.start(EQC.Pulse.GenEvent, mode, @no_callback, [], [])
      atom when is_atom(atom) ->
        :gen.start(EQC.Pulse.GenEvent, mode, {:local, atom}, @no_callback, [], [])
      other when is_tuple(other) ->
        :gen.start(EQC.Pulse.GenEvent, mode, other, @no_callback, [], [])
    end
  end

  @doc """
  Returns a stream that consumes events from the `manager`.

  The stream is a `GenEvent` struct that implements the `Enumerable`
  protocol. Consumption of events only begins when enumeration starts.

  Note streaming is specific to Elixir's GenEvent and does not work
  with Erlang ones.

  ## Options

    * `:timeout` - raises if no event arrives in X milliseconds
      (defaults to `:infinity`)

  """
  @spec stream(manager, Keyword.t) :: GenEvent.Stream.t
  def stream(manager, options \\ []) do
    %GenEvent.Stream{
      manager: manager,
      timeout: Keyword.get(options, :timeout, :infinity)}
  end

  @doc """
  Adds a new event handler to the event `manager`.

  The event manager will call the `init/1` callback with `args` to
  initiate the event handler and its internal state.

  If `init/1` returns a correct value indicating successful completion,
  the event manager adds the event handler and this function returns
  `:ok`. If the callback fails with `reason` or returns `{:error, reason}`,
  the event handler is ignored and this function returns `{:error, reason}`.

  If the given handler was previously installed at the manager, this
  function returns `{:error, :already_present}`.
  """
  @spec add_handler(manager, handler, term) :: :ok | {:error, term}
  def add_handler(manager, handler, args) do
    rpc(manager, {:add_handler, handler, args})
  end

  @doc """
  Adds a monitored event handler to the event `manager`.

  Expects the same input and returns the same values as `add_handler/3`.

  ## Monitored handlers

  A monitored handler implies the calling process will now be monitored
  by the GenEvent manager.

  If the calling process later terminates with `reason`, the event manager
  will delete the event handler by calling the `terminate/2` callback with
  `{:stop, reason}` as argument. If the event handler later is deleted,
  the event manager sends a message `{:gen_event_EXIT, handler, reason}`
  to the calling process. Reason is one of the following:

    * `:normal` - if the event handler has been removed due to a call to
      `remove_handler/3`, or `:remove_handler` has been returned by a callback
      function

    * `:shutdown` - if the event handler has been removed because the event
      manager is terminating

    * `{:swapped, new_handler, pid}` - if the process pid has replaced the
      event handler by another

    * a term - if the event handler is removed due to an error. Which term
      depends on the error

  Keep in mind that the `{:gen_event_EXIT, handler, reason}` message is not
  guaranteed to be delivered in case the manager crashes. If you want to
  guarantee the message is delivered, you have two options:

    * monitor the event manager
    * link to the event manager and then set `Process.flag(:trap_exit, true)`
      in your handler callback

  Finally, this functionality only works with GenEvent started via this
  module (it is not backwards compatible with Erlang's `:gen_event`).
  """
  @spec add_mon_handler(manager, handler, term) :: :ok | {:error, term}
  def add_mon_handler(manager, handler, args) do
    rpc(manager, {:add_mon_handler, handler, args, self()})
  end

  @doc """
  Sends an event notification to the event `manager`.

  The event manager will call `handle_event/2` for each
  installed event handler.

  `notify` is asynchronous and will return immediately after the
  notification is sent. `notify` will not fail even if the specified
  event manager does not exist, unless it is specified as an atom.
  """
  @spec notify(manager, term) :: :ok
  def notify(manager, event)

  def notify({:global, name}, msg) do
    try do
      :global.send(name, {:notify, msg})
      :ok
    catch
      _, _ -> :ok
    end
  end

  def notify({:via, mod, name}, msg) do
    try do
      mod.send(name, {:notify, msg})
      :ok
    catch
      _, _ -> :ok
    end
  end

  def notify(other, msg) do
    send(other, {:notify, msg})
    :ok
  end

  @doc """
  Sends a sync event notification to the event `manager`.

  In other words, this function only returns `:ok` after the event manager
  invokes the `handle_event/2` on each installed event handler.

  See `notify/2` for more info.
  """
  @spec sync_notify(manager, term) :: :ok
  def sync_notify(manager, event) do
    rpc(manager, {:sync_notify, event})
  end

  @doc """
  Sends a ack event notification to the event `manager`.

  In other words, this function only returns `:ok` as soon as the
  event manager starts processing this event, but it does not wait
  for event handlers to process the sent event.

  See `notify/2` for more info. Note this function is specific
  to Elixir's GenEvent and does not work with Erlang ones.
  """
  @spec ack_notify(manager, term) :: :ok
  def ack_notify(manager, event) do
    rpc(manager, {:ack_notify, event})
  end

  @doc """
  Makes a synchronous call to the event `handler` installed in `manager`.

  The given `request` is sent and the caller waits until a reply arrives or
  a timeout occurs. The event manager will call `handle_call/2` to handle
  the request.

  The return value `reply` is defined in the return value of `handle_call/2`.
  If the specified event handler is not installed, the function returns
  `{:error, :not_found}`.
  """
  @spec call(manager, handler, term, timeout) ::  term | {:error, term}
  def call(manager, handler, request, timeout \\ 5000) do
    try do
      :gen.call(manager, self(), {:call, handler, request}, timeout)
    catch
      :exit, reason ->
        exit({reason, {__MODULE__, :call, [manager, handler, request, timeout]}})
    else
      {:ok, res} -> res
    end
  end

  @doc """
  Removes an event handler from the event `manager`.

  The event manager will call `terminate/2` to terminate the event handler
  and return the callback value. If the specified event handler is not
  installed, the function returns `{:error, :not_found}`.
  """
  @spec remove_handler(manager, handler, term) :: term | {:error, term}
  def remove_handler(manager, handler, args) do
    rpc(manager, {:delete_handler, handler, args})
  end

  @doc """
  Replaces an old event handler with a new one in the event `manager`.

  First, the old event handler is deleted by calling `terminate/2` with
  the given `args1` and collects the return value. Then the new event handler
  is added and initiated by calling `init({args2, state}), where term is the
  return value of calling `terminate/2` in the old handler. This makes it
  possible to transfer information from one handler to another.

  The new handler will be added even if the specified old event handler
  is not installed or if the handler fails to terminate with a given reason
  in which case `state = {:error, term}`.

  If `init/1` in the second handler returns a correct value, this
  function returns `:ok`.
  """
  @spec swap_handler(manager, handler, term, handler, term) :: :ok | {:error, term}
  def swap_handler(manager, handler1, args1, handler2, args2) do
    rpc(manager, {:swap_handler, handler1, args1, handler2, args2})
  end

  @doc """
  Replaces an old event handler with a new monitored one in the event `manager`.

  Read the docs for `add_mon_handler/3` and `swap_handler/5` for more information.
  """
  @spec swap_mon_handler(manager, handler, term, handler, term) :: :ok | {:error, term}
  def swap_mon_handler(manager, handler1, args1, handler2, args2) do
    rpc(manager, {:swap_mon_handler, handler1, args1, handler2, args2, self()})
  end

  @doc """
  Returns a list of all event handlers installed in the `manager`.
  """
  @spec which_handlers(manager) :: [handler]
  def which_handlers(manager) do
    rpc(manager, :which_handlers)
  end

  @doc """
  Terminates the event `manager`.

  Before terminating, the event manager will call `terminate(:stop, ...)`
  for each installed event handler.
  """
  @spec stop(manager) :: :ok
  def stop(manager) do
    rpc(manager, :stop)
  end

  defp rpc(module, cmd) do
    # TODO: Change the tag once patch is accepted by OTP
    {:ok, reply} = :gen.call(module, self(), cmd, :infinity)
    reply
  end

  ## Init callbacks

  require Record
  Record.defrecordp :handler, [:module, :id, :state, :pid, :ref]

  @doc false
  def init_it(starter, :self, name, mod, args, options) do
    init_it(starter, self(), name, mod, args, options)
  end

  def init_it(starter, parent, name, _, _, options) do
    Process.put(:"$initial_call", {__MODULE__, :init_it, 6})
    debug = :gen.debug_options(options)
    :proc_lib.init_ack(starter, {:ok, self()})
    loop(parent, name(name), [], debug, false)
  end

  @doc false
  def init_hib(parent, name, handlers, debug) do
    fetch_msg(parent, name, handlers, debug, true)
  end

  defp name({:local, name}),  do: name
  defp name({:global, name}), do: name
  defp name({:via, _, name}), do: name
  defp name(pid) when is_pid(pid), do: pid

  ## Loop

  defp loop(parent, name, handlers, debug, true) do
    :proc_lib.hibernate(__MODULE__, :init_hib, [parent, name, handlers, debug])
  end

  defp loop(parent, name, handlers, debug, false) do
    fetch_msg(parent, name, handlers, debug, false)
  end

  defp fetch_msg(parent, name, handlers, debug, hib) do
    receive do
      {:system, from, req} ->
        :sys.handle_system_msg(req, from, parent, __MODULE__,
          debug, [name, handlers, hib], hib)
      {:EXIT, ^parent, reason} ->
        server_terminate(reason, parent, handlers, name)
      msg when debug == [] ->
        handle_msg(msg, parent, name, handlers, [])
      msg ->
        debug = :sys.handle_debug(debug, &print_event/3, name, {:in, msg})
        handle_msg(msg, parent, name, handlers, debug)
    end
  end

  defp handle_msg(msg, parent, name, handlers, debug) do
    case msg do
      {:notify, event} ->
        {hib, handlers} = server_event(:async, event, handlers, name)
        loop(parent, name, handlers, debug, hib)
      {_from, _tag, {:notify, event}} ->
        {hib, handlers} = server_event(:async, event, handlers, name)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:ack_notify, event}} ->
        reply(tag, :ok)
        {hib, handlers} = server_event(:ack, event, handlers, name)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:sync_notify, event}} ->
        {hib, handlers} = server_event(:sync, event, handlers, name)
        reply(tag, :ok)
        loop(parent, name, handlers, debug, hib)
      {:DOWN, ref, :process, _pid, reason} = other ->
        case handle_down(ref, reason, handlers, name) do
          {:ok, handlers} ->
            loop(parent, name, handlers, debug, false)
          :error ->
            {hib, handlers} = server_info(other, handlers, name)
            loop(parent, name, handlers, debug, hib)
        end
      {_from, tag, {:call, handler, query}} ->
        {hib, reply, handlers} = server_call(handler, query, handlers, name)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:add_handler, handler, args}} ->
        {hib, reply, handlers} = server_add_handler(handler, args, handlers)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:add_mon_handler, handler, args, notify}} ->
        {hib, reply, handlers} = server_add_mon_handler(handler, args, handlers, notify)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:add_process_handler, pid, notify}} ->
        {hib, reply, handlers} = server_add_process_handler(pid, handlers, notify)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:delete_handler, handler, args}} ->
        {reply, handlers} = server_remove_handler(handler, args, handlers, name)
        reply(tag, reply)
        loop(parent, name, handlers, debug, false)
      {_from, tag, {:swap_handler, handler1, args1, handler2, args2}} ->
        {hib, reply, handlers} = server_swap_handler(handler1, args1, handler2, args2, handlers, nil, name)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, {:swap_mon_handler, handler1, args1, handler2, args2, mon}} ->
        {hib, reply, handlers} = server_swap_handler(handler1, args1, handler2, args2, handlers, mon, name)
        reply(tag, reply)
        loop(parent, name, handlers, debug, hib)
      {_from, tag, :stop} ->
        try do
          server_terminate(:normal, parent, handlers, name)
        catch
          :exit, :normal -> :ok
        end
        reply(tag, :ok)
      {_from, tag, :which_handlers} ->
        reply(tag, server_which_handlers(handlers))
        loop(parent, name, handlers, debug, false)
      {_from, tag, :get_modules} ->
        reply(tag, server_get_modules(handlers))
        loop(parent, name, handlers, debug, false)
      other ->
        {hib, handlers} = server_info(other, handlers, name)
        loop(parent, name, handlers, debug, hib)
    end
  end

  ## System callbacks

  @doc false
  def system_continue(parent, debug, [name, handlers, hib]) do
    loop(parent, name, handlers, debug, hib)
  end

  @doc false
  def system_terminate(reason, parent, _debug, [name, handlers, _hib]) do
    server_terminate(reason, parent, handlers, name)
  end

  @doc false
  def system_code_change([name, handlers, hib], module , old_vsn, extra) do
    handlers =
      for handler <- handlers do
        if handler(handler, :module) == module do
          {:ok, state} = module.code_change(old_vsn, handler(handler, :state), extra)
          handler(handler, state: state)
        else
          handler
        end
      end
    {:ok, [name, handlers, hib]}
  end

  @doc false
  def system_get_state([_name, handlers, _hib]) do
    tuples = for handler(module: mod, id: id, state: state) <- handlers do
      {mod, id, state}
    end
    {:ok, tuples}
  end

  @doc false
  def system_replace_state(fun, [name, handlers, hib]) do
    {handlers, states} =
      :lists.unzip(for handler <- handlers do
        handler(module: mod, id: id, state: state) = handler
        cur = {mod, id, state}
        try do
          new = {^mod, ^id, new_state} = fun.(cur)
          {handler(handler, state: new_state), new}
        catch
          _, _ ->
            {handler, cur}
        end
      end)
    {:ok, states, [name, handlers, hib]}
  end

  @doc false
  def format_status(opt, status_data) do
    [pdict, sys_state, parent, _debug, [name, handlers, _hib]] = status_data
    header = :gen.format_status_header('Status for event handler', name)

    formatted = for handler <- handlers do
      handler(module: module, state: state) = handler
      if function_exported?(module, :format_status, 2) do
        try do
          state = module.format_status(opt, [pdict, state])
          handler(handler, state: state)
        catch
          _, _ -> handler
        end
      else
        handler
      end
    end

    [header: header,
     data: [{'Status', sys_state}, {'Parent', parent}],
     items: {'Installed handlers', formatted}]
  end

  ## Loop helpers

  defp print_event(dev, {:in, msg}, name) do
    case msg do
      {:notify, event} ->
        IO.puts dev, "*DBG* #{inspect name} got event #{inspect event}"
      {_, _, {:call, handler, query}} ->
        IO.puts dev, "*DBG* #{inspect name} (handler #{inspect handler}) got call #{inspect query}"
      _ ->
        IO.puts dev, "*DBG* #{inspect name} got #{inspect msg}"
    end
  end

  defp print_event(dev, dbg, name) do
    IO.puts dev, "*DBG* #{inspect name}: #{inspect dbg}"
  end

  defp server_add_handler({module, id}, args, handlers) do
    handler = handler(module: module, id: {module, id})
    do_add_handler(module, handler, args, handlers, :ok)
  end

  defp server_add_handler(module, args, handlers) do
    handler = handler(module: module, id: module)
    do_add_handler(module, handler, args, handlers, :ok)
  end

  defp server_add_mon_handler({module, id}, args, handlers, notify) do
    ref = Process.monitor(notify)
    handler = handler(module: module, id: {module, id}, pid: notify, ref: ref)
    do_add_handler(module, handler, args, handlers, :ok)
  end

  defp server_add_mon_handler(module, args, handlers, notify) do
    ref = Process.monitor(notify)
    handler = handler(module: module, id: module, pid: notify, ref: ref)
    do_add_handler(module, handler, args, handlers, :ok)
  end

  defp server_add_process_handler(pid, handlers, notify) do
    ref = Process.monitor(pid)
    handler = handler(module: GenEvent.Stream, id: {self(), ref},
                      pid: notify, ref: ref)
    do_add_handler(GenEvent.Stream, handler, {pid, ref}, handlers, {self(), ref})
  end

  defp server_remove_handler(module, args, handlers, name) do
    do_take_handler(module, args, handlers, name, :remove, :normal)
  end

  defp server_swap_handler(module1, args1, module2, args2, handlers, sup, name) do
    {state, handlers} =
      do_take_handler(module1, args1, handlers, name, :swapped, {:swapped, module2, sup})

    if sup do
      server_add_mon_handler(module2, {args2, state}, handlers, sup)
    else
      server_add_handler(module2, {args2, state}, handlers)
    end
  end

  defp server_info(event, handlers, name) do
    handlers = :lists.reverse(handlers)
    server_notify(event, :handle_info, handlers, name, handlers, [], false)
  end

  defp server_event(mode, event, handlers, name) do
    {handlers, streams} = server_split_process_handlers(mode, event, handlers, [], [])
    {hib, handlers} = server_notify(event, :handle_event, handlers, name, handlers, [], false)
    {hib, server_collect_process_handlers(mode, event, streams, handlers, name)}
  end

  defp server_split_process_handlers(mode, event, [handler|t], handlers, streams) do
    case handler(handler, :id) do
      {pid, _ref} when is_pid(pid) ->
        server_process_notify(mode, event, handler)
        server_split_process_handlers(mode, event, t, handlers, [handler|streams])
      _ ->
        server_split_process_handlers(mode, event, t, [handler|handlers], streams)
    end
  end

  defp server_split_process_handlers(_mode, _event, [], handlers, streams) do
    {handlers, streams}
  end

  defp server_process_notify(mode, event, handler(state: {pid, ref})) do
    send pid, {self(), {self(), ref}, {mode_to_tag(mode), event}}
  end

  defp mode_to_tag(:ack),   do: :ack_notify
  defp mode_to_tag(:sync),  do: :sync_notify
  defp mode_to_tag(:async), do: :notify

  defp server_notify(event, fun, [handler|t], name, handlers, acc, hib) do
    case server_update(handler, fun, event, name, handlers) do
      {new_hib, handler} ->
        server_notify(event, fun, t, name, handlers, [handler|acc], hib or new_hib)
      :error ->
        server_notify(event, fun, t, name, handlers, acc, hib)
    end
  end

  defp server_notify(_, _, [], _, _, acc, hib) do
    {hib, acc}
  end

  defp server_update(handler, fun, event, name, _handlers) do
    handler(module: module, state: state) = handler

    case do_handler(module, fun, [event, state]) do
      {:ok, res} ->
        case res do
          {:ok, state} ->
            {false, handler(handler, state: state)}
          {:ok, state, :hibernate} ->
            {true, handler(handler, state: state)}
          :remove_handler ->
            do_terminate(handler, :remove_handler, event, name, :normal)
            :error
          other ->
            reason = {:bad_return_value, other}
            do_terminate(handler, {:error, reason}, event, name, reason)
            :error
        end
      {:error, reason} ->
        do_terminate(handler, {:error, reason}, event, name, reason)
        :error
    end
  end

  defp server_collect_process_handlers(:async, event, [handler|t], handlers, name) do
    server_collect_process_handlers(:async, event, t, [handler|handlers], name)
  end

  defp server_collect_process_handlers(mode, event, [handler|t], handlers, name) when mode in [:sync, :ack] do
    handler(ref: ref, id: id) = handler

    receive do
      {^ref, :ok} ->
        server_collect_process_handlers(mode, event, t, [handler|handlers], name)
      {_from, tag, {:delete_handler, ^id, args}} ->
        do_terminate(handler, args, :remove, name, :normal)
        reply(tag, :ok)
        server_collect_process_handlers(mode, event, t, handlers, name)
      {:DOWN, ^ref, _, _, reason} ->
        do_terminate(handler, {:stop, reason}, :DOWN, name, :shutdown)
        server_collect_process_handlers(mode, event, t, handlers, name)
    end
  end

  defp server_collect_process_handlers(_mode, _event, [], handlers, _name) do
    handlers
  end

  defp server_call(module, query, handlers, name) do
    case :lists.keyfind(module, handler(:id) + 1, handlers) do
      false ->
        {false, {:error, :not_found}, handlers}
      handler ->
        case server_call_update(handler, query, name, handlers) do
          {{hib, handler}, reply} ->
            {hib, reply, :lists.keyreplace(module, handler(:id) + 1, handlers, handler)}
          {:error, reply} ->
            {false, reply, :lists.keydelete(module, handler(:id) + 1, handlers)}
        end
    end
  end

  defp server_call_update(handler, query, name, _handlers) do
    handler(module: module, state: state) = handler
    case do_handler(module, :handle_call, [query, state]) do
      {:ok, res} ->
        case res do
          {:ok, reply, state} ->
            {{false, handler(handler, state: state)}, reply}
          {:ok, reply, state, :hibernate} ->
            {{true, handler(handler, state: state)}, reply}
          {:remove_handler, reply} ->
            do_terminate(handler, :remove_handler, query, name, :normal)
            {:error, reply}
          other ->
            reason = {:bad_return_value, other}
            do_terminate(handler, {:error, reason}, query, name, reason)
            {:error, {:error, reason}}
        end
      {:error, reason} ->
        do_terminate(handler, {:error, reason}, query, name, reason)
        {:error, {:error, reason}}
    end
  end

  defp server_get_modules(handlers) do
    (for handler(module: module) <- handlers, do: module)
    |> :ordsets.from_list
    |> :ordsets.to_list
  end

  defp server_which_handlers(handlers) do
    for handler(id: id) <- handlers, do: id
  end

  defp server_terminate(reason, _parent, handlers, name) do
    _ =
      for handler <- handlers do
        do_terminate(handler, :stop, :stop, name, :shutdown)
      end
    exit(reason)
  end

  defp reply({from, ref}, msg) do
    send from, {ref, msg}
  end

  defp handle_down(ref, reason, handlers, name) do
    case :lists.keyfind(ref, handler(:ref) + 1, handlers) do
      false -> :error
      handler ->
        do_terminate(handler, {:stop, reason}, :DOWN, name, :shutdown)
        {:ok, :lists.keydelete(ref, handler(:ref) + 1, handlers)}
    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_present}, handlers}
    end
  end

  defp do_take_handler(module, args, handlers, name, last_in, reason) do
    case :lists.keytake(module, handler(:id) + 1, handlers) do
      {:value, handler, handlers} ->
        {do_terminate(handler, args, last_in, name, reason), handlers}
      false ->
        {{:error, :not_found}, handlers}
    end
  end

  defp do_terminate(handler, arg, last_in, name, reason) do
    handler(module: module, state: state) = handler

    res =
      case do_handler(module, :terminate, [arg, state]) do
        {:ok, res} -> res
        {:error, _} = error -> error
      end
    report_terminate(handler, reason, state, last_in, name)
    res
  end

  defp do_handler(mod, fun, args) do
    try do
      apply(mod, fun, args)
    catch
      :throw, val -> {:ok, val}
      :error, val -> {:error, {val, System.stacktrace}}
      :exit, val  -> {:error, val}
    else
      res -> {:ok, res}
    end
  end

  defp report_terminate(handler, reason, state, last_in, name) do
    report_error(handler, reason, state, last_in, name)
    if ref = handler(handler, :ref) do
      Process.demonitor(ref, [:flush])
    end
    if pid = handler(handler, :pid) do
      send pid, {:gen_event_EXIT, handler(handler, :id), reason}
    end
  end

  defp report_error(_handler, :normal, _, _, _), do: :ok
  defp report_error(_handler, :shutdown, _, _, _), do: :ok
  defp report_error(_handler, {:swapped, _, _}, _, _, _), do: :ok
  defp report_error(handler, reason, state, last_in, name) do
    reason =
      case reason do
        {:undef, [{m,f,a,_}|_]=mfas} ->
          cond do
            :code.is_loaded(m) ->
              {:"module could not be loaded", mfas}
            function_exported?(m, f, length(a)) ->
              reason
            true ->
              {:"function not exported", mfas}
          end
        _ ->
          reason
      end

    formatted = report_status(handler, state)

    :error_logger.error_msg(
      '** gen_event handler ~p crashed.~n' ++
      '** Was installed in ~p~n' ++
      '** Last event was: ~p~n' ++
      '** When handler state == ~p~n' ++
      '** Reason == ~p~n', [handler(handler, :id), name, last_in, formatted, reason])
  end

  defp report_status(handler(module: module), state) do
    if function_exported?(module, :format_status, 2) do
      try do
        module.format_status(:terminate, [Process.get(), state])
      catch
        _, _ -> state
      end
    else
      state
    end
  end
end