Elixir: How to consume a stream of server-sent events as a client?

user_1729692 0
Viewed 386

My Phoenix application needs to integrate with a 3rd-party API that sends data in a stream format via Server-Sent Events. I'm learning about SSE for the first time; SSE streams seem a lot like websockets except they're unidirectional and way simpler.

I see plenty of advice on how to send SSE data from an Elixir app as the server (for ex. 1, 2, 3) - but I see no advice on how an Elixir application can consume a stream of SSE data as a client. The articles I saw seem to assume that the client is JS in the browser, but I know that Ruby SSE clients exist (1, 2) so I assume there's no strong technical reason why an Elixir app wouldn't be able to act as a SSE client.

What's the easiest way to consume SSE data in an Elixir app? I'd be grateful for any pointers & resources.

server-sent-events elixir
1 Answers

I found a couple promising SSE client libraries, but I got turned off by weird connection errors. Then after some digging, I discovered that HTTPoison (and presumably also HTTPotion) can connect to a Server-Sent Event stream perfectly fine on its own, it just isn't documented anywhere.

Here's a quick iex example (using Elixir v1.8 and HTTPoison v1.5):

> url = "https://some-domain.com/some-server-sent-event-stream/"
> HTTPoison.get!(url, [], [recv_timeout: :infinity, stream_to: self()])
  => %HTTPoison.AsyncResponse{id: #Reference<0.2736682462.4075814917.25838>}
> Process.sleep(10_000) # wait for a few events to come in
> flush()
  => %HTTPoison.AsyncChunk{
        chunk: "event:poke\ndata:{\"kittens\":3}\n\n",
        id: #Reference<0.2736682462.4075814917.25838>
      }
      %HTTPoison.AsyncChunk{
        chunk: "event:poke\ndata:{\"kittens\":3}\n\n",
        id: #Reference<0.2736682462.4075814917.25838>
      }
      %HTTPoison.AsyncChunk{
        chunk: "event:poke\ndata:{\"kittens\":4}\n\n",
        id: #Reference<0.2736682462.4075814917.25838>
      }
      :ok

But in the real world, you'll likely want to use a GenServer instead to process each message, handle disconnects, etc. -- and this is essentially what the two above-linked libraries were meant to do. A simple client might look like this:

# Usage:
# > SseClient.start("https://some-domain.com/some-server-sent-event-stream/")
#
defmodule SseClient do
  use GenServer

  def start(url) do
    GenServer.start_link(__MODULE__, url: url)
  end

  def init([url: url]) do
    IO.puts "Connecting to stream..."
    HTTPoison.get!(url, [], [recv_timeout: :infinity, stream_to: self()])
    {:ok, nil}
  end

  def handle_info(%HTTPoison.AsyncChunk{chunk: chunk}, _state) do
    # My use case assumes that each message contains two rows (event: and data:)
    case Regex.run(~r/^event:(\w+)\ndata:({.+})\n\n$/, chunk) do
      [_, event, data] ->
        _json = Jason.decode!(data)
        case event do
          "poke" -> IO.puts "Poke received: #{data}"
          "data" -> IO.puts "Data received: #{data}"
        end
      nil ->
        raise "Don't know how to parse received chunk: \"#{chunk}\""
    end

    {:noreply, nil}
  end

  # In addition to message chunks, we also may receive status changes etc.
  def handle_info(%HTTPoison.AsyncStatus{} = status, _state) do
    IO.puts "Connection status: #{inspect status}"
    {:noreply, nil}
  end

  def handle_info(%HTTPoison.AsyncHeaders{} = headers, _state) do
    IO.puts "Connection headers: #{inspect headers}"
    {:noreply, nil}
  end
end
user_17296920
Related