Streams and collection updates

Prerequisites

Read HTML and event bindings before choosing a stream over ordinary keyed rendering. The entries rendered by this guide must have stable domain keys.

Choose Streams Deliberately

Ordinary Scala collections are the default. Render a collection normally when the complete value belongs in the model and ordinary tree diffing is sufficient. Use splitBy to give repeated entries stable keys. Keyed rendering sends sparse entry updates and positional references, then lets the client patch the resulting HTML normally.

Use a stream when collection changes should instead become explicit ID-addressed insert, update, delete, or reset operations in the browser. Streams are especially useful for feeds, logs, and bounded windows that change frequently. They require more explicit ownership than keyed rendering, so they are not a general replacement for Vector, database state, or another durable source of truth.

Separate Domain State From Stream State

A LiveStream[A] is an opaque, immutable rendering handle. It intentionally does not expose collection operations or its entries. Keep queryable domain data separately and store the latest stream handle beside it:

final case class Model(
  activities: Vector[Activity],
  activityStream: LiveStream[Activity],
  nextId: Int
)

The Vector can answer application questions such as total activity count and category totals. The stream handle carries the current snapshot and pending DOM operations. Stream state belongs to one socket or component lifecycle, so mount must recreate it after a remount.

This separation also permits a bounded DOM without discarding domain data. The activity example keeps its complete history in a Vector while retaining only five rendered rows.

Define Stable Identity And Retention

Create one LiveStreamDef[A, Id] for each logical stream. Its name identifies the stream within the owning LiveView or component, and its DOM-ID function identifies rows:

private val ActivityStreamDef =
  LiveStreamDef.byId[Activity, Int]("activity")(_.id).keepLast(5)

Every generated ID must be stable, non-empty, and unique in the rendered document. Inserting an item with an existing ID updates that row in place. Changing an item's ID does not remove the row with its old ID; delete the old identity explicitly when a domain operation changes identity.

keepFirst(count) and keepLast(count) apply a retention policy during create, reset, and insertion operations. Counts must be positive. Retention limits the stream snapshot and rendered DOM, not separately retained domain state.

Create The Stream During Mount

Call ctx.streams.create once for a stream name in each lifecycle and retain the returned handle:

def mount(ctx: MountContext): Task[Model] =
  ctx.streams.create(ActivityStreamDef, InitialActivities).map { stream =>
    Model(InitialActivities, stream, nextId = 5)
  }

Creating the same name twice for one owner fails. Definitions are owner-scoped, but rendered container and row IDs are still document IDs and must remain unique across nested LiveViews and component instances.

Retain Every Replacement Handle

Every stream operation returns a replacement handle. Store and render that exact value; rendering an older handle loses the pending operation:

case Msg.Add =>
  val activity = Activity(model.nextId, "Streams", "Inserted one row")
  ctx.streams.insert(ActivityStreamDef, activity).map { stream =>
    model.copy(
      activities = model.activities :+ activity,
      activityStream = stream,
      nextId = model.nextId + 1
    )
  }

insert appends by default. Supply a StreamAt value to insert first or at an index, and use updateOnly = true to ignore a missing identity instead of inserting it. Bulk insertion behaves like repeated insertion at the same position, so repeatedly inserting first or at one fixed index can reverse input order.

Delete And Reset Coherently

Apply the same domain operation to durable data and the stream. The definition retains the domain ID type and maps that ID to the row's DOM ID, so deletion does not need the complete item:

case Msg.Delete(activityId) =>
  ctx.streams.delete(ActivityStreamDef, activityId).map { stream =>
    model.copy(
      activities = model.activities.filterNot(_.id == activityId),
      activityStream = stream
    )
  }

deleteByDomId is a lower-level alternative for a trusted rendered DOM ID belonging to that stream. Prefer delete with a typed domain ID, and do not pass untrusted browser input to the DOM-ID operation.

reset replaces the stream snapshot and instructs the browser to rebuild the container. Reset durable state in the same message handler when the user-facing contract resets the whole example:

case Msg.Reset =>
  ctx.streams.reset(ActivityStreamDef, InitialActivities).map { stream =>
    Model(InitialActivities, stream, nextId = 5)
  }

Render The Stream Container

renderIn creates the stream container, assigns its required update mode, and assigns each projected row its generated DOM ID:

model.activityStream.renderIn(ol, aria.label := "Recent activity") { activity =>
  li(
    p(activity.summary),
    button(on.click(Msg.Delete(activity.id)), "Delete")
  )
}

Do not override the container or row IDs produced by the stream. Keep controls inside each projected row typed to the owning LiveView's message type as usual.

The complete implementation is extracted from executable source:

Source

final class ActivityStreamExample
    extends LiveView[ActivityStreamExample.Msg, ActivityStreamExample.Model]:
  import ActivityStreamExample.*

  def mount(ctx: MountContext): Task[Model] =
    initialModel(ctx)

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.Add =>
      val template = NewActivityTemplates((model.nextId - 1) % NewActivityTemplates.size)
      val activity = Activity(model.nextId, template._1, template._2)
      ctx.streams.insert(ActivityStreamDef, activity).map { stream =>
        model.copy(
          activities = model.activities :+ activity,
          activityStream = stream,
          nextId = model.nextId + 1
        )
      }
    case Msg.Delete(activityId) =>
      ctx.streams.delete(ActivityStreamDef, activityId).map { stream =>
        model.copy(
          activities = model.activities.filterNot(_.id == activityId),
          activityStream = stream
        )
      }
    case Msg.Reset =>
      ctx.streams.reset(ActivityStreamDef, InitialActivities).map { stream =>
        Model(InitialActivities, stream, InitialNextId)
      }

  override def view(model: Signal[Model]): HtmlElement[Msg] =
    div(
      cls := "docs-activity-stream",
      sectionTag(
        cls        := "docs-activity-summary",
        aria.label := "Activity stream state",
        div(
          span(cls := "docs-activity-label", "Durable history"),
          strong(
            cls                        := "docs-activity-count",
            dataAttr("activity-count") := "",
            model.map(_.activities.size.toString)
          )
        ),
        p(
          span("DOM window"),
          "Five recent rows. Complete history remains in the model."
        )
      ),
      div(
        cls := "docs-activity-controls",
        button(
          typ                      := "button",
          dataAttr("add-activity") := "",
          on.click(Msg.Add),
          "Insert activity"
        )
      ),
      model
        .map(_.activityStream).renderIn(
          ol,
          cls        := "docs-activity-list",
          aria.label := "Recent activity"
        ) { activity =>
          li(
            dataAttr("activity-row") := "",
            div(
              cls := "docs-activity-row-content",
              span(cls  := "docs-activity-category", activity.map(_.category)),
              p(cls     := "docs-activity-message", activity.map(_.summary)),
              small(cls := "docs-activity-id", activity.map(value => s"Activity #${value.id}"))
            ),
            button(
              cls                         := "docs-activity-delete",
              typ                         := "button",
              dataAttr("delete-activity") := activity.map(_.id.toString),
              aria.label                  := activity.map(value => s"Delete activity ${value.id}"),
              on.click(activity.map(value => Msg.Delete(value.id))),
              "Delete"
            )
          )
        }
    )

  private def initialModel(ctx: MountContext): Task[Model] =
    ctx.streams.create(ActivityStreamDef, InitialActivities).map { stream =>
      Model(InitialActivities, stream, InitialNextId)
    }
end ActivityStreamExample

object ActivityStreamExample:
  final case class Activity(id: Int, category: String, summary: String)

  final case class Model(
    activities: Vector[Activity],
    activityStream: LiveStream[Activity],
    nextId: Int)

  enum Msg:
    case Add
    case Delete(activityId: Int)
    case Reset

  private val ActivityStreamDef =
    LiveStreamDef.byId[Activity, Int]("activity")(_.id).keepLast(5)

  private val InitialActivities = Vector(
    Activity(1, "Navigation", "Opened the typed search example"),
    Activity(2, "Forms", "Validated a profile draft"),
    Activity(3, "Rendering", "Applied a keyed collection patch"),
    Activity(4, "Components", "Updated component props by stable ID")
  )

  private val InitialNextId = InitialActivities.map(_.id).max + 1

  private val NewActivityTemplates = Vector(
    "Streams"    -> "Inserted a bounded activity row",
    "Navigation" -> "Patched to another results page",
    "Components" -> "Updated component props by stable ID",
    "Async work" -> "Completed a deterministic report"
  )
end ActivityStreamExample

View source (documentation/site/src/scalive/docs/examples/ActivityStreamExample.scala:8-129)

Try insertion, bounded retention, deletion, and reset in the bounded activity stream example.

Keep simpler repeated content keyed with HTML and event bindings.
Feed repeated updates from lifecycle-owned work with Asynchronous work and subscriptions.
Exercise initial collection rendering with Testing LiveViews.