Stampede.SSE (stampede v1.4.0)

Copy Markdown View Source

A client for Server-Sent Events, the streams that AI chat APIs, live feeds and many dashboards send their updates on.

{:ok, response} = Stampede.SSE.get("http://localhost:4000/api/updates")
[%{event: "message", data: data, at: at} | _] = response.events

Requests go through Stampede.HTTP, so each VU keeps its connection to a server open between them, as a browser does. The response has the :status, the :headers, the raw :body, and the :events read from it, in order. Each event has:

  • :event - its type, "message" unless the server set another
  • :data - its data, with the lines of a multi-line event joined by "\n"
  • :id - the last id the server sent, or nil
  • :at - when it arrived, in milliseconds from sending the request

Events follow the HTML standard's rules for text/event-stream. An event the stream ended in the middle of, without the blank line after it, still counts, since some servers end that way.

Metrics

Inside a run, requests record:

  • "sse.request" - from sending the request to the end of the stream
  • "sse.first_event" - from sending the request to the first event
  • "sse.events" - how many events arrived
  • "sse.request.errors" - requests that failed, or whose response had a status of 400 or more

Summary

Types

event()
response()

Functions

get(url, opts \\ [])

Opens a stream with a GET request. See request/3.

post(url, opts \\ [])

Opens a stream with a POST request. See request/3.

request(method, url, opts \\ [])

Sends a method request to url and reads the events it streams back, until the server ends the stream.

Types

event()

@type event() :: %{
  event: String.t(),
  data: String.t(),
  id: String.t() | nil,
  at: float()
}

response()

@type response() :: %{
  status: non_neg_integer(),
  headers: [{String.t(), String.t()}],
  body: binary(),
  events: [event()]
}

Functions

get(url, opts \\ [])

@spec get(String.t(), keyword()) :: {:ok, response()} | {:error, term()}

Opens a stream with a GET request. See request/3.

post(url, opts \\ [])

@spec post(String.t(), keyword()) :: {:ok, response()} | {:error, term()}

Opens a stream with a POST request. See request/3.

request(method, url, opts \\ [])

@spec request(atom(), String.t(), keyword()) :: {:ok, response()} | {:error, term()}

Sends a method request to url and reads the events it streams back, until the server ends the stream.

Returns {:ok, response} whatever the status, and {:error, reason} when the stream failed, such as a Mint.TransportError when the connection closed before the end.

Options

  • :headers - a list of {name, value} headers. An accept header of text/event-stream is added unless one is given.
  • :body - the body, as iodata
  • :json - a term to send as JSON instead of a :body
  • :cookies - when true, uses the cookies of Stampede.HTTP's :cookies option, so a stream can come after signing in. Defaults to false.
  • :timeout - how long the whole stream may take, in milliseconds. Defaults to 60_000.
  • :metric - the prefix of the metrics to record, or nil to record nothing. Defaults to "sse".
  • :transport_opts - options for :gen_tcp or :ssl, given to Mint.HTTP.connect/4