A gen_statem-style event loop that fits the Plug contract.

Elixir HTTP servers using Plug require every write to a Plug.Conn to come from the process that received the request. PlugStatem.enter_loop/6 takes over that process during the execution of a long-lived response, such as when serving SSE.

Over the lifetime of the response, your callback module will be executed just like a standard gen_statem.

On :stop, PlugStatem.enter_loop/5 returns a result instead of exiting the process, so that the rest of the Plug call/2 contract can be fulfilled.

There are a handful of message shapes that are reserved by Bandit. Messages that match these patterns will not be delivered to your module as an :info event.

• {:bandit, _}
• {:plug_conn, :sent}

Example

A Plug's call/2 enters the loop and returns the conn when finished:

def call(%Plug.Conn{path_info: ["events"]} = conn, _opts) do
  conn = Plug.Conn.put_resp_content_type(conn, "text/event-stream")

  case PlugStatem.enter_loop(MyStream, [], conn, :starting, %{}) do
    {_reason, _state, conn, _data} -> conn
  end
end

Your module streams chunks to the conn:

defmodule MyStream do
  @behaviour PlugStatem

  @keep_alive {{:timeout, :keep_alive}, 15_000, nil}

  @impl true
  def callback_mode, do: [:handle_event_function, :state_enter]

  @impl true
  def handle_event(:enter, _old, :streaming, conn, data) do
    {:keep_state, Plug.Conn.send_chunked(conn, 200), data, @keep_alive}
  end

  def handle_event(:info, {:event, iodata}, :streaming, conn, data) do
    write(conn, iodata, data)
  end

  def handle_event({:timeout, :keep_alive}, nil, :streaming, conn, data) do
    write(conn, ": keep-alive\n\n", data)
  end

  def handle_event(:info, :stop, :streaming, conn, data) do
    {:stop, :normal, conn, data}
  end

  defp write(conn, iodata, data) do
    case Plug.Conn.chunk(conn, iodata) do
      {:ok, conn} -> {:keep_state, conn, data, @keep_alive}
      {:error, reason} -> {:stop, {:shutdown, reason}, conn, data}
    end
  end
end

Other processes can drive the stream by sending it messages, for example send(pid, {:event, "data: hello\n\n"}). With the :name option they can find it through a Registry instead of holding the pid.

Callbacks

The callback module mirrors gen_statem, with the Plug.Conn threaded through every callback alongside state and data.

callback_mode() :: mode | [mode | :state_enter]
  where mode :: :handle_event_function | :state_functions

# callback_mode :handle_event_function
handle_event(event_type, event_content, state, conn, data) :: result

# callback_mode :state_functions, one function per state atom
state_name(event_type, event_content, conn, data) :: result

# optional; see enter_loop/4
init(init_arg, conn) :: {:ok, state, conn, data}
                      | {:ok, state, conn, data, actions}
                      | {:stop, reason, conn}

# optional
terminate(reason, state, conn, data) :: term

Events

:info                message that is not one of the below
{:call, from}        from `:gen_statem.call/2,3`; reply with a `:reply` action
:cast                from `:gen_statem.cast/2`
:state_timeout       content of a `:state_timeout` action
:timeout             content of a `:timeout` action
{:timeout, name}     content of a `{:timeout, name}` action
:enter               state enter call; content is the previous state
any other type       inserted by a `:next_event` action

Results

{:next_state, state, conn, data}
{:next_state, state, conn, data, actions}
{:keep_state, conn, data}
{:keep_state, conn, data, actions}
:keep_state_and_data
{:keep_state_and_data, actions}
{:repeat_state, conn, data}
{:repeat_state, conn, data, actions}
:repeat_state_and_data
{:repeat_state_and_data, actions}
{:stop, reason}
{:stop, reason, conn, data}
{:stop_and_reply, reason, replies}
{:stop_and_reply, reason, replies, conn, data}

A state enter call may not change the state and may not use :postpone or :next_event.

Actions

{:reply, from, reply}
:postpone | {:postpone, boolean}
{:next_event, event_type, event_content}
{:state_timeout, ms | :infinity, content}
{:state_timeout, :cancel}
{:timeout, ms | :infinity, content}
{:timeout, :cancel}
{{:timeout, name}, ms | :infinity, content}
{{:timeout, name}, :cancel}

Not yet implemented, and rejected with an ArgumentError naming the action:

:hibernate
{timeout_type, :update, content}
{timeout_type, ms, content, opts}
{:change_callback_module, module}
{:push_callback_module, module}
:pop_callback_module

format_status/1 and code_change/4 are never called.

After a state change, events are handled in this order: events inserted by :next_event, then postponed events in the order they were postponed, then anything already queued, then the mailbox.

Options

enter_loop/6 takes a keyword list of options as its second argument, in the position GenServer.start_link/3 uses:

• :name - registers the current process for the lifetime of the loop under a name, in any of the forms GenServer accepts: an atom, a {:global, term}, or a {:via, module, term} such as {:via, Registry, {MyRegistry, key}}. The name is unregistered when the loop returns, including when a callback raises. If the name is already taken, enter_loop/6 returns {:error, {:already_started, pid}} without running any callback.