Notify chunked connection pid of EventSource client closing connection (or crashing)

[zampino]

With the current Plug.Adapters.Cowboy I am not able to react on tcp socket being closed
during chunked connections. Could be that I am doing something wrong, but it's maybe worth
to start a discussion on this.

Typical scenario: I register, monitor and 'simple_one_for_one'-supervise a
GenEvent process for each incoming event source connection:

defmodule Streaming.Router do
  use Plug.Router
  plug :match
  plug :dispatch

  def init(_), do: []

  get "/connections/:id" do
    put_resp_content_type(conn, "text/event-stream")
    |> send_chunked(200)
    |> register_and_link_events_manager(id)
    |> listen
  end

  defp listen(conn) do
    GenEvent.stream(conn.assigns.events_manager_pid)
    |> Stream.each(&chunk(conn, format(&1)))
    |> Stream.run
  end

Since I link the registered process, when the connection closes
it would be enough to send a (non normal) exit signal to the process and let
my gen_server do the necessary cleanup. But I couldn't find a way to do that.

As far as I understand, to get this kind of feedback the actual tcp_socket must be marked as active.

Transport:setopts(Socket, [{active, once}])

This for instance happens in a cowboy handler which enters the loop behaviour in
its init phase. Actual cowboy handler uses the upgrade protocol.

On the other hand Elli server has a nice loop-like
response handler (which is also mentioned in #65). To be more precise,
based on the user-configured handler, the server decides to enter or not the
chunked loop based on the
current middleware environment and this sounds perfectly plug-ish.

I drafted an Elli adapter: Pastelli with a subset of the Plug.Conn adapter interface, focusing on the 'chunked' functionalities. I would be of course very happy to
contribute to plug itself if a built-in Elli adapter is planned.

Elli response model supports a.o.
handler callbacks on several "events" like "reponse_complete", "request_parse_error" and in particular for 'chunk_complete' which is what I need.

For a streaming scenario like the one above, Pastelli doesn't block the action dispatch
(which happens above by calling GenEvent.stream) but yield the connection to
the server which enters a receive loop waiting
for 'chunk' messages. Finally, passing the conn struct to your event handler we can send chunks to the client with the usual api.

[josevalim]

Thank you! To be clear, you need a way to know if the connection is closed. It seems that Elli allows you to register something and, that something, is going to trigger an exit signal in your case.

For this to work with cowboy, it seems our best option is to mark the socket as active in another process that watches the process with the connection and kills it once it notices tcp is closed. Is this correct?

For a streaming scenario like the one above, Pastelli doesn't block the action dispatch (which happens above by calling GenEvent.stream) but yield the connection to the server which enters a receive loop waiting for 'chunk' messages.

Isn't Stream.run/1 going to block this current process receiving messages and writing them to client? Do you mean there is another process responsible for controlling the socket?

[zampino]

• Not sure it's a good practice to mark the socket as active for all requests, cowboy code has comments
  about it and when it's better to mark it back to passive if I am not wrong...
• Sure the GenEvent.stream + run example above works for the cowboy adapter only because
  in that case I have to block. If I remember well the upgrade call would otherwise send
  the response back to the client and that would close the eventsource connection.
  In case of a Router served by Pastelli you would write an action as described in https://github.com/zampino/pastelli/blob/master/README.md
• my heroku deploy was unfortunate today :-( but I hope to paste some working example here soon 💃

[fishcakez]

It would be possible to create a built in plug such as: Plug.Task.call(conn, {plug, args}) that works exactly like the chunk_loop.

[zampino]

@josevalim

or this to work with cowboy, it seems our best option is to mark the socket as active in another process that watches the process with the connection

ok maybe I get your idea: setting up a chunk_complete callback directly in the
middleware. This would be great (btw the EventMachine/Thin/Sinatra stack already do this :-)

In a router:

get "/connections/:id" do
    put_resp_content_type(conn, "text/event-stream")
    |> send_chunked(200)
    |> on_chunk_complete(fn(owner, socket)-> Process.exit(owner) end) # but something else 
    |> listen
  end

in Plug.Conn
we can spawn a new process and "steal" the controlling process of the
socket and mark it as active there:

def on_chunk_complete(%Conn{state: :chunked, adapter: {handler, req}} = conn, callback) do
  conn  = %Conn{conn | chunk_callback: callback}
  spawn_link fn ->
    [socket, transport] = cowboy_req:get([:socket, :transport], req)
    :gen_tcp.controlling_process(socket, self)
    transport.setopts(socket, [{:active, :once}])
    receive do
      {tcp_closed, socket} -> conn.chunk_callback.(conn.owner, socket)
      # and possibly also:
      {:tcp_error, socket, reason}-> conn.errback(conn.owner, socket)
    end
  end
  conn
end

this would be nice, although we are bringing
the upgrade protocol closer to a loop handler and we don't know
the side effects of activating the socket.

Nevertheless worth trying it out, probably.

[josevalim]

@zampino that's what I had in mind but, after looking at both cowboy and elli implementation, it seems supporting some sort of chunk_loop would be simpler and more efficient, even if we have to implement most of the chunk loop ourselves and simply ask the adapters to provide a mechanism for hijacking the socket (which we would need for websockets anyway).

So I think a hijacking API on the adapter side is probably best. Thoughts?

[zampino]

@josevalim

we have to implement most of the chunk loop ourselves

I am not sure..... what if we tell cowboy to enter the :loop behavior as expected
and let it handle socket communication himself, then we still keep the present chunk api on Plug side?

The only part there is that to pass the HandlerState we should resolve our plug.call at an
earlier stage (in Plug handler's init probably?) so that we know the intention ("sent" or "chunked") of the middleware.
Also look at the terminate_handler callback in cowboy.

With Elli this "intention"-switch is relatively easier, but what I
still need to understand is how/when to pass (and get back) the whole connection struct in order to
react on events with a finer behavior instead of -- in the chunk_complete case -- a semi-brutal exit(:shutdown).

By the way cowboy 2 will probably handle this stuff differently, are there plans to
react on a new cowboy release?

[josevalim]

I am not sure..... what if we tell cowboy to enter the :loop behavior as expected
and let it handle socket communication himself, then we still keep the present chunk api on Plug side?

The downside of this approach is that we are still dependent on the cowboy API and the decisions it has made. If we can hijack the socket and implement our own loop, our loop would work with elli, cowboy and what not. The adapter logic becomes then the hijacking not the whole loop itself. That is the same rationale for websockets. If we hijack, we can also provide websockets for all adapters.

By the way cowboy 2 will probably handle this stuff differently, are there plans to
react on a new cowboy release?

We can update whenever there is a release and someone provides a pull request. :)

[tudborg]

@zampino Did you do any work on this?

[zampino]

@tbug check this elli plug adapter I've been working on Pastelli
the README should explain related issues.

I guess any work toward cowboy 2 will obsolete this (or has already?) :-) @josevalim ?

[tudborg]

@zampino Already had a look.
Looks great.
Functionally it's what I need, I had just hoped to stay on cowboy. 😄
But my project is not that Cowboy specific, so I guess I could give Elli a go.

[lilactown]

I'm trying to use server-sent events using plug + cowboy. Any solutions for this?