Skip to content
scalive
Menu
ConnectingLiveReconnectingOffline
GitHub

Subscriptions

Before you start

Start with a connected LiveView whose messages already produce explicit model transitions. Loading data and live updates introduces subscriptions with a complete example. Use asynchronous work instead for one finite task.

A subscription runs a ZStream[Any, Nothing, Msg] owned by one connected LiveView or component and sends each emitted value to handleMessage. Scalive stops it when its owner closes. A ZStream[R, E, A] emits zero or more A values over time; the subscription's stream needs no environment and cannot fail.

Start a subscription when the page connects

Most pages subscribe in mount, once the browser has connected, and stay subscribed while the page is open. Match ctx.connection and start the stream in the connected branch:

Source
scala
def mount(ctx: MountContext): Task[Model] =
  ctx.connection match
    case Connection.Disconnected =>
      ZIO.succeed(Model(reading = AsyncValue.loading))
    case Connection.Connected(connected) =>
      for
        _ <- connected.async.start(Reading)(thermostat.temperature)(Msg.Measured(_))
        _ <- connected.subscriptions.start(RefreshTimer, SubscriptionDelivery.Latest)(
               ZStream.fromSchedule(Schedule.spaced(RefreshEvery)).as(Msg.Refresh)
             )
      yield Model(reading = AsyncValue.loading)

View source (documentation/fixtures/learn/src/quickstart/ThermostatLiveView.scala:14-24)

connected.subscriptions exists only in the Connection.Connected branch, so the disconnected HTTP render cannot start a stream. Every connected mount, including the one after a reconnect, subscribes again. This stream emits Msg.Refresh every ten seconds.

When a service announces changes itself, subscribe to its stream rather than polling, and map each change to a message:

Fragment
scala
connected.subscriptions.start(Changes, SubscriptionDelivery.Lossless)(
  orders.placed.map(Msg.OrderPlaced(_))
)

Loading a snapshot and subscribing to its changes separately can miss or reorder a change made while both are starting. Prefer a service stream that emits the current value first and then each change, so one subscription keeps the model current.

After mount, message handlers reach the same capability through ctx.subscriptions.

Own long-lived streams with subscriptions

A SubscriptionKey identifies one registered stream in its root LiveView or exact component owner. The stream must emit that owner's Msg type and cannot fail. The clock example keys its subscription by instance, SubscriptionKey(s"subscription-clock-$instanceId"), and builds its stream with:

Source
scala
private def ticks(every: Duration): ZStream[Any, Nothing, Msg] =
  ZStream.tick(every).mapZIO(_ => Clock.instant).map(Msg.Tick(_))

View source (documentation/site/src/scalive/docs/examples/SubscriptionClockExample.scala:98-99)

start rejects a duplicate registered key, including a component registration whose worker is suspended by dormancy. Use it when starting twice indicates a state-machine mistake. The clock guards start with its model so the button and server transition agree:

Source
scala
case Msg.Start =>
  if model.mode == Mode.Stopped then
    ctx.subscriptions
      .start(ClockSubscription, SubscriptionDelivery.Lossless)(ticks(1.second))
      .as(model.copy(mode = Mode.EverySecond))
  else ZIO.succeed(model)

View source (documentation/site/src/scalive/docs/examples/SubscriptionClockExample.scala:22-27)

Replace or cancel a subscription

replace starts or swaps the stream under a key. Replacing the registration interrupts the old stream and resubscribes the runtime's current set. Use replacement when changing polling frequency, topic, props, or another stream parameter is valid application behavior:

Source
scala
case Msg.Replace =>
  ctx.subscriptions
    .replace(ClockSubscription, SubscriptionDelivery.Lossless)(ticks(250.millis))
    .as(model.copy(mode = Mode.FourTimesPerSecond))

View source (documentation/site/src/scalive/docs/examples/SubscriptionClockExample.scala:30-33)

cancel removes the registration and succeeds when it is already absent. Update the model in the same handler so the rendered controls describe the registered state:

Source
scala
case Msg.Cancel =>
  ctx.subscriptions.cancel(ClockSubscription).as(model.copy(mode = Mode.Stopped))

View source (documentation/site/src/scalive/docs/examples/SubscriptionClockExample.scala:36-37)

Choose a delivery policy

The required SubscriptionDelivery argument controls backpressure at the LiveView mailbox. Lossless delivers every emitted message in order; use it when every transition matters. Latest coalesces pending delivery so newer messages supersede older ones; use it for high-frequency state snapshots or refresh signals where only the newest value matters. The clock deliberately uses Lossless for both start and replacement so every tick is counted.

Root subscription messages pass through the root's typed info lifecycle hooks before handleMessage. Registrations exist only for the connected lifecycle and do not run during disconnected rendering. Mount therefore starts required streams again for each new connected lifecycle.

Construct each stream so resources are acquired anew every time it runs. Put handles inside ZStream.scoped or ZStream.unwrapScoped; do not acquire a handle first and capture it in a stream value. Replacement and component revival rerun the stored stream, so a captured handle may already have been released.

Own a stream in a component

Components expose the same Subscriptions[Msg] capability through connected.subscriptions during connected mount and update, and through ctx.subscriptions while handling a component message. A registered stream emits the component's Msg, not the parent LiveView's message type. Delivery follows the component's typed event hooks and handleMessage. Root info hooks do not handle component messages.

The subscription namespace is the exact runtime component instance. Two instances can deliberately reuse the same SubscriptionKey without collision, and root registrations remain independent. A disconnected component mount or update cannot acquire a subscription because the capability exists only in its Connection.Connected branch.

When a rendered component becomes dormant while the runtime waits for browser removal confirmation, Scalive stops its subscription worker but retains the registration with the component model. If that same instance returns first, the runtime runs the stored ZStream again with fresh internal tokens, without calling mount or update merely to restart it. Pending values from the old worker are discarded. This is interruption and resubscription, not a replay or lossless handoff guarantee.

Only a stream still actively registered at dormancy is eligible for this automatic restart. An explicitly cancelled stream, a normally completed stream, or a defective stream stays stopped. Once the browser confirms destruction, Scalive removes the registration; rendering the same class and logical ID later mounts a fresh component and may register a fresh stream. Root subscriptions are unaffected by component dormancy.

Keep ownership local

Keys are runtime identities, not persistence keys. Keep them stable within one owner and unique among that owner's subscriptions. Deriving root keys from instance identity makes ownership visible in traces and prevents collisions when code moves into a shared owner. On page exit, Scalive stops every registered stream; on remount, start them again from the page's real source of truth.

Study the complete examples

The clock implementation shows start, replacement, cancellation, reset, and an instance-scoped subscription key:

Source
scala
final class SubscriptionClockExample(instanceId: String)
    extends LiveView[SubscriptionClockExample.Msg, SubscriptionClockExample.Model]:
  import SubscriptionClockExample.*

  private val ClockSubscription = subscriptionKey(instanceId)

  def mount(ctx: MountContext): Task[Model] =
    ZIO.succeed(Model())

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.Start =>
      if model.mode == Mode.Stopped then
        ctx.subscriptions
          .start(ClockSubscription, SubscriptionDelivery.Lossless)(ticks(1.second))
          .as(model.copy(mode = Mode.EverySecond))
      else ZIO.succeed(model)
    case Msg.Replace =>
      ctx.subscriptions
        .replace(ClockSubscription, SubscriptionDelivery.Lossless)(ticks(250.millis))
        .as(model.copy(mode = Mode.FourTimesPerSecond))
    case Msg.Cancel =>
      ctx.subscriptions.cancel(ClockSubscription).as(model.copy(mode = Mode.Stopped))
    case Msg.Reset =>
      ctx.subscriptions.cancel(ClockSubscription).as(Model())
    case Msg.Tick(at) =>
      ZIO.succeed(model.copy(lastTick = Some(at), tickCount = model.tickCount + 1))

  def view(model: Signal[Model]): HtmlElement[Msg] =
    div(
      cls := "docs-managed-work",
      sectionTag(
        cls        := "docs-managed-work-state",
        aria.label := "Clock subscription state",
        p("Mode", strong(dataAttr("clock-mode") := "", model.map(_.mode.label))),
        p("Ticks received", strong(dataAttr("clock-count") := "", model.map(_.tickCount.toString))),
        p(
          "Latest tick",
          span(dataAttr("clock-tick") := "", model.map(_.lastTick.fold("Waiting")(_.toString)))
        )
      ),
      div(
        cls := "docs-managed-work-controls",
        button(
          typ      := "button",
          disabled := model.map(_.mode != Mode.Stopped),
          on.click(Msg.Start),
          "Start every second"
        ),
        button(typ := "button", on.click(Msg.Replace), "Replace with fast clock"),
        button(
          typ      := "button",
          disabled := model.map(_.mode == Mode.Stopped),
          on.click(Msg.Cancel),
          "Cancel clock"
        )
      )
    )
end SubscriptionClockExample

object SubscriptionClockExample:
  enum Mode(val label: String):
    case Stopped            extends Mode("Stopped")
    case EverySecond        extends Mode("Every second")
    case FourTimesPerSecond extends Mode("Four times per second")

  final case class Model(
    mode: Mode = Mode.Stopped,
    lastTick: Option[Instant] = None,
    tickCount: Int = 0)

  enum Msg:
    case Start
    case Replace
    case Cancel
    case Reset
    case Tick(at: Instant)

  private[docs] def subscriptionKey(instanceId: String): SubscriptionKey =
    SubscriptionKey(s"subscription-clock-$instanceId")

  private def ticks(every: Duration): ZStream[Any, Nothing, Msg] =
    ZStream.tick(every).mapZIO(_ => Clock.instant).map(Msg.Tick(_))

View source (documentation/site/src/scalive/docs/examples/SubscriptionClockExample.scala:11-100)

The component implementation shows two instances safely reusing one local key, mount-started streams, local replacement and cancellation, and parent-controlled visibility:

Source
scala
object SubscriptionTickerComponent
    extends LiveComponent[
      SubscriptionTickerComponent.Props,
      SubscriptionTickerComponent.Msg,
      SubscriptionTickerComponent.Model
    ]:
  final case class Props(id: String, title: String, resetEpoch: Int)

  enum Mode(val label: String):
    case Stopped            extends Mode("Stopped")
    case EverySecond        extends Mode("Every second")
    case FourTimesPerSecond extends Mode("Four times per second")

  final case class Model(
    mode: Mode = Mode.EverySecond,
    ticks: Int = 0,
    resetEpoch: Int = 0)

  enum Msg:
    case Start
    case Replace
    case Cancel
    case Tick

  // Both component instances deliberately reuse this key. The runtime namespace
  // is the exact component instance, so their registrations do not collide.
  private val LocalTicks = SubscriptionKey("component-local-ticks")

  def mount(props: Props, ctx: MountContext): Task[Model] =
    ctx.connection match
      case Connection.Disconnected         => ZIO.succeed(Model(resetEpoch = props.resetEpoch))
      case Connection.Connected(connected) =>
        connected.subscriptions
          .start(LocalTicks, SubscriptionDelivery.Lossless)(ticks(1.second))
          .as(Model(resetEpoch = props.resetEpoch))

  override def update(props: Props, model: Model, ctx: UpdateContext): Task[Model] =
    if props.resetEpoch == model.resetEpoch then ZIO.succeed(model)
    else
      ctx.connection match
        case Connection.Disconnected         => ZIO.succeed(Model(resetEpoch = props.resetEpoch))
        case Connection.Connected(connected) =>
          connected.subscriptions
            .replace(LocalTicks, SubscriptionDelivery.Lossless)(ticks(1.second))
            .as(Model(resetEpoch = props.resetEpoch))

  def handleMessage(props: Props, model: Model, ctx: MessageContext) =
    case Msg.Start =>
      if model.mode == Mode.Stopped then
        ctx.subscriptions
          .start(LocalTicks, SubscriptionDelivery.Lossless)(ticks(1.second))
          .as(model.copy(mode = Mode.EverySecond))
      else ZIO.succeed(model)
    case Msg.Replace =>
      ctx.subscriptions
        .replace(LocalTicks, SubscriptionDelivery.Lossless)(ticks(250.millis))
        .as(model.copy(mode = Mode.FourTimesPerSecond))
    case Msg.Cancel =>
      ctx.subscriptions.cancel(LocalTicks).as(model.copy(mode = Mode.Stopped))
    case Msg.Tick =>
      ZIO.succeed(model.copy(ticks = model.ticks + 1))

  def view(props: Signal[Props], model: Signal[Model], self: ComponentRef[Msg]) =
    articleTag(
      cls                                := "docs-vote-card",
      dataAttr("subscription-component") := props.map(_.id),
      headerTag(
        div(
          p(cls := "docs-vote-kicker", "Component-owned subscription"),
          h4(props.map(_.title))
        ),
        code(dataAttr("component-id") := props.map(_.id), props.map(_.id))
      ),
      div(
        cls := "docs-vote-count-row",
        div(
          cls := "docs-vote-metric",
          span("Ticks"),
          strong(dataAttr("component-ticks") := "", model.map(_.ticks.toString))
        ),
        p(dataAttr("component-mode") := "", model.map(_.mode.label))
      ),
      div(
        cls := "docs-vote-actions",
        button(
          typ      := "button",
          disabled := model.map(_.mode != Mode.Stopped),
          phx.target(self),
          on.click.to(self)(Msg.Start),
          "Start local ticks"
        ),
        button(
          typ := "button",
          phx.target(self),
          on.click.to(self)(Msg.Replace),
          "Replace local ticks"
        ),
        button(
          typ      := "button",
          disabled := model.map(_.mode == Mode.Stopped),
          phx.target(self),
          on.click.to(self)(Msg.Cancel),
          "Cancel local ticks"
        )
      )
    )

  private def ticks(every: Duration): ZStream[Any, Nothing, Msg] =
    ZStream.repeatZIO(ZIO.sleep(every).as(Msg.Tick))
end SubscriptionTickerComponent

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

  def mount(ctx: MountContext): Task[Model] = ZIO.succeed(Model())

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.ToggleFirst => ZIO.succeed(model.copy(firstVisible = !model.firstVisible))
    case Msg.Reset       =>
      ZIO.succeed(model.copy(firstVisible = true, resetEpoch = model.resetEpoch + 1))

  def view(model: Signal[Model]): HtmlElement[Msg] =
    div(
      cls := "docs-voting-components",
      sectionTag(
        cls        := "docs-vote-parent",
        aria.label := "Parent component visibility",
        div(
          p(cls := "docs-vote-kicker", "Parent-owned visibility"),
          p(
            dataAttr("first-visibility") := "",
            model.map(value =>
              if value.firstVisible then "First ticker is rendered."
              else "First ticker is removed."
            )
          )
        ),
        button(
          typ := "button",
          on.click(Msg.ToggleFirst),
          model.map(value =>
            if value.firstVisible then "Remove first ticker" else "Reinsert first ticker"
          )
        )
      ),
      div(
        cls := "docs-vote-grid",
        model
          .map(_.firstVisible).when(
            div(
              FirstTicker.render(
                model.map(value =>
                  SubscriptionTickerComponent
                    .Props("first-ticker", "First ticker", value.resetEpoch)
                )
              )
            )
          ),
        SecondTicker.render(
          model.map(value =>
            SubscriptionTickerComponent.Props("second-ticker", "Second ticker", value.resetEpoch)
          )
        )
      )
    )
end ComponentSubscriptionsExample

object ComponentSubscriptionsExample:
  final case class Model(firstVisible: Boolean = true, resetEpoch: Int = 0)

  enum Msg:
    case ToggleFirst
    case Reset

  private val FirstTicker  = component(SubscriptionTickerComponent, "first-ticker")
  private val SecondTicker = component(SubscriptionTickerComponent, "second-ticker")

View source (documentation/site/src/scalive/docs/examples/ComponentSubscriptionsExample.scala:9-185)

Run the managed clock subscription and component-owned subscriptions alongside the documentation site's diagnostic views to correlate messages, model transitions, and final DOM changes.