brazier/stream

Streams: a keyed list of items the server sends once and does not keep (docs/design/STREAMS.md). A chat log, a feed or an agent’s output grows without the model holding every item: update returns ops that insert, update, delete or reset items, the runtime sends the change, and the client keeps the items, at most the stream’s limit of them.

Declare each stream once, in a pub fn, so update and view share it:

import brazier/stream.{type Stream}
import gleam/dynamic/decode
import gleam/json

pub type Line {
  Line(id: String, text: String)
}

pub fn log() -> Stream(Line) {
  stream.new(
    "log",
    key: fn(line: Line) { line.id },
    encode: fn(line: Line) {
      json.object([
        #("id", json.string(line.id)),
        #("text", json.string(line.text)),
      ])
    },
    decoder: {
      use id <- decode.field("id", decode.string)
      use text <- decode.field("text", decode.string)
      decode.success(Line(id:, text:))
    },
    limit: stream.KeepLast(300),
  )
}

Each op calls the stream’s key and encode when it is called, so it holds the item’s key and JSON text, never the item.

Each op function returns an Effect(msg). Ops compose with effect.batch, in order, and effect.map passes them through, so a child module’s ops reach its parent. effect.keyed and effect.cancellable do not key them: an op is applied when its update returns. emitted lists the ops an effect carries, for tests.

render puts the stream in the view: a container element whose children are the stream’s items, each rendered by a named top-level function of the item (docs/design/STREAMS.md 5.3):

import brazier/element.{type Element}
import brazier/html
import brazier/stream.{type Stream}

pub type Line {
  Line(id: String, text: String)
}

pub fn view(log: Stream(Line)) -> Element(msg) {
  html.main([], [stream.render(log, "ol", [], render_line)])
}

pub fn render_line(line: Line) -> Element(msg) {
  html.li([], [html.text(line.text)])
}

Types

Emitted

</>

One stream op, for app unit tests: the stream’s name, the item’s key and its encoded JSON text.

pub type Emitted {
  Inserted(
    stream: String,
    key: String,
    json: String,
    at: Position,
  )
  Deleted(stream: String, key: String)
  Reset(stream: String, items: List(#(String, String)))
}

Constructors

Inserted(stream: String, key: String, json: String, at: Position)

```
Deleted(stream: String, key: String)
```

Reset(stream: String, items: List(#(String, String)))

Limit

</>

Which end a full stream keeps. With KeepLast(300) an append past 300 items drops the oldest one at the front (LiveView limit: -300); with KeepFirst(300) a prepend past 300 drops the last one (LiveView limit: 300). An append into a full KeepFirst stream, or a prepend into a full KeepLast one, drops the new item itself.

pub type Limit {
  KeepFirst(Int)
  KeepLast(Int)
}

Constructors

```
KeepFirst(Int)
```
```
KeepLast(Int)
```

Position

</>

Where a new item goes. An item whose key is in the stream stays where it is and is updated in place.

pub type Position {
  Append
  Prepend
}

Constructors

```
Append
```
```
Prepend
```

Stream

opaque </>

A keyed stream of items of one type, made by new.

pub opaque type Stream(item)

Values

delete

</>

pub fn delete(
  stream: Stream(item),
  item: item,
) -> effect.Effect(msg)

Delete the item with item’s key. A key not in the stream changes nothing; the runtime logs it at warning.

delete_key

</>

pub fn delete_key(
  stream: Stream(item),
  key: String,
) -> effect.Effect(msg)

Delete the item with key (LiveView stream_delete_by_dom_id/3).

emitted

</>

pub fn emitted(effect: effect.Effect(msg)) -> List(Emitted)

The stream ops an effect carries, in the order they are applied: an insert_many with Prepend lists its items from the last. The effect’s other callbacks are ignored.

stream.emitted(stream.delete_key(log(), "7"))
// -> [stream.Deleted(stream: "log", key: "7")]

insert

</>

pub fn insert(
  stream: Stream(item),
  item: item,
  at position: Position,
) -> effect.Effect(msg)

Insert item at position, or update it in place when its key is in the stream.

stream.insert(log(), Line(id: "7", text: "done"), at: stream.Append)

insert_many

</>

pub fn insert_many(
  stream: Stream(item),
  items: List(item),
  at position: Position,
) -> effect.Effect(msg)

Insert several items: Append keeps the list’s order at the end, Prepend puts the list’s first item first. When a key appears twice in items, it takes the place of its first appearance and the item of its last, as one insert.

stream.insert_many(log(), [first, second], at: stream.Prepend)
// The stream now starts with first, then second.

key

</>

pub fn key(stream: Stream(item), item: item) -> String

The key the stream gives item.

limit

</>

pub fn limit(stream: Stream(item)) -> Limit

The stream’s limit.

name

</>

pub fn name(stream: Stream(item)) -> String

The stream’s name.

new

</>

pub fn new(
  name: String,
  key key: fn(item) -> String,
  encode encode: fn(item) -> json.Json,
  decoder decoder: decode.Decoder(item),
  limit limit: Limit,
) -> Stream(item)

Declare a stream. The name identifies it on the wire and in the page.

Crashes, as attribute.attribute does for a bad name, when name is not 1 to 64 characters of a-z, 0-9 and _ starting with a letter, or when the limit is below 1. The largest limit is a server bound (SecurityLimits.stream_max_limit, 1000 by default): a stream over it has its ops refused and logged by the runtime.

stream.new(
  "log",
  key: fn(line: Line) { line.id },
  encode: encode_line,
  decoder: line_decoder(),
  limit: stream.KeepLast(300),
)

render

</>

pub fn render(
  stream: Stream(item),
  tag: String,
  attributes: List(attribute.Attribute(msg)),
  item: fn(item) -> element.Element(msg),
) -> element.Element(msg)

The stream’s container: a tag element holding the stream’s items, in order, each rendered by item with its root keyed by the stream’s key (LiveView’s phx-update="stream" container).

item must be a named top-level function, not an anonymous fn or a capture: an item is rendered when it is inserted or updated, never again, so it can depend on the item only. Its root must be one element, not text, element.none(), unsafe_raw_html or element.keyed. A view renders a stream at most once and never inside another stream’s item; a second render of the same stream crashes the render. The build checks the rest (docs/design/STREAMS.md 5.3).

stream.render(log(), "ol", [attribute.class("log")], render_line)

reset

</>

pub fn reset(
  stream: Stream(item),
  items: List(item),
) -> effect.Effect(msg)

Empty the stream, then insert items in order at the end. reset(stream, []) empties it. A key that appears twice is one item, as in insert_many.

round_trip

</>

pub fn round_trip(
  stream: Stream(item),
  item: item,
) -> Result(item, String)

Encode item and decode it again with the stream’s decoder, so a test can check the stream’s codec. The error describes the decode failure.