howdy/sse

Server-sent events: a response that stays open so the server can push messages to a browser as things happen.

An event stream is an ordinary GET route, so it goes through the same middleware, guards and versioning as any other. The handler describes the stream and finishes with start.

import howdy/sse

controller.new("orders")
|> controller.get("/events", fn(ctx) {
  sse.new(fn(stream) {
    sse.join(stream, "orders")
    Nil
  })
  |> sse.start(ctx)
})

and from anywhere else in the program:

sse.broadcast("orders", sse.json(order_json) |> sse.name("order.created"))

In a browser, new EventSource("/orders/events") connects to it and reconnects by itself if the connection drops, sending back the id of the last event it saw; see last_event_id.

Server-sent events only go one way. When the client needs to talk back, it sends ordinary requests, or the app uses howdy/websocket instead.

Each stream runs in its own process, like a WebSocket. Callbacks are given the Stream to send on and the current state, and return what to do next: continue with new state, or stop. Other processes reach a stream through subject, and groups of streams through topics with join and broadcast.

Types

Builder

opaque </>

A stream description under construction. state is what the callbacks carry between messages; msg is what subject accepts.

pub opaque type Builder(state, msg)

Event

</>

A message on the stream. Build one with event or json, then set the other fields with name, id and retry.

pub type Event {
  Event(
    data: String,
    name: option.Option(String),
    id: option.Option(String),
    retry: option.Option(Int),
  )
}

Constructors

```
Event(
  data: String,
  name: option.Option(String),
  id: option.Option(String),
  retry: option.Option(Int),
)
```
Arguments

data

What the client receives as the event’s data. Text spanning several lines arrives as it was sent.

name

The event’s type. Browsers deliver named events to listeners added with addEventListener(name, ...), and unnamed ones to onmessage.

id

Sent back by a reconnecting client in the last-event-id header.

retry

How long, in milliseconds, the client waits before reconnecting.

opaque </>

What a callback asks the stream to do once it is done.

pub opaque type Next(state)

Stream

opaque </>

A handle for one open stream. msg is the type of message other processes may send it; see subject.

pub opaque type Stream(msg)

Values

broadcast

</>

pub fn broadcast(topic: String, event: Event) -> Nil

Send an event to every stream on a topic, on every connected node. Each stream sends it from its own process, so this returns at once and never waits on a slow client.

continue

</>

pub fn continue(state: state) -> Next(state)

Keep the stream open and handle the next message with state.

event

</>

pub fn event(data: String) -> Event

An unnamed event carrying data.

heartbeat

</>

pub fn heartbeat(
  builder: Builder(state, msg),
  milliseconds: Int,
) -> Builder(state, msg)

Send a comment every milliseconds to keep an idle stream open. A heartbeat that cannot be sent ends the stream, which is also how a client that went away without a word is noticed.

id

</>

pub fn id(event: Event, id: String) -> Event

Give an event an ID. A client that reconnects sends the last ID it saw, so the stream can pick up where it left off.

join

</>

pub fn join(stream: Stream(msg), topic: String) -> Nil

Subscribe a stream to a topic. Joining a topic twice has no extra effect. Membership ends when the stream does.

json

</>

pub fn json(body: json.Json) -> Event

An unnamed event carrying body as JSON.

last_event_id

</>

pub fn last_event_id(
  stream: Stream(msg),
) -> option.Option(String)

The ID of the last event the client saw before it reconnected, from the last-event-id header. None on a first connection.

leave

</>

pub fn leave(stream: Stream(msg), topic: String) -> Nil

Unsubscribe a stream from a topic. Leaving a topic the stream is not on does nothing.

name

</>

pub fn name(event: Event, name: String) -> Event

Give an event a type, for the client’s addEventListener.

new

</>

pub fn new(
  on_open: fn(Stream(msg)) -> state,
) -> Builder(state, msg)

Describe a stream. on_open runs once the response headers have been sent and returns the starting state. It may send events straight away, such as the ones a reconnecting client missed.

A comment is sent every 15 seconds so that proxies do not close an idle stream; see heartbeat.

no_heartbeat

</>

pub fn no_heartbeat(
  builder: Builder(state, msg),
) -> Builder(state, msg)

Send no heartbeat.

on_close

</>

pub fn on_close(
  builder: Builder(state, msg),
  handler: fn(Stream(msg), state) -> Nil,
) -> Builder(state, msg)

Run once when the stream ends, however that happens. Nothing can be sent to the client from here.

on_message

</>

pub fn on_message(
  builder: Builder(state, msg),
  handler: fn(Stream(msg), state, msg) -> Next(state),
) -> Builder(state, msg)

Handle messages sent to the stream’s subject by other processes.

retry

</>

pub fn retry(event: Event, milliseconds: Int) -> Event

Tell the client how long to wait, in milliseconds, before reconnecting if the connection drops.

send

</>

pub fn send(
  stream: Stream(msg),
  event: Event,
) -> Result(Nil, content.SendError)

Send an event to this stream’s client.

send_comment

</>

pub fn send_comment(
  stream: Stream(msg),
  text: String,
) -> Result(Nil, content.SendError)

Send a comment, which clients ignore. The heartbeat is one of these.

size

</>

pub fn size(topic: String) -> Int

How many streams are on a topic.

start

</>

pub fn start(
  builder: Builder(state, msg),
  ctx: context.Context(guarded),
) -> response.Response(content.Content)

Respond with the stream. The response is 200 with the text/event-stream content type, and the stream then runs until a callback returns stop, the client goes away or the server shuts down.

stop

</>

pub fn stop() -> Next(state)

End the stream. on_close still runs. A browser’s EventSource reconnects after a stream ends; to stop it for good, close it from the client, or answer the reconnect with a status other than 200, such as 204.

subject

</>

pub fn subject(stream: Stream(msg)) -> process.Subject(msg)

A subject other processes can send to. Each message arrives at the on_message callback. Hand it to whatever needs to push to this client.

Constructors

Arguments