Skip to content
scalive
Menu
ConnectingLiveReconnectingOffline

Asynchronous work and subscriptions

Prerequisites

Start with Scalive's model and message lifecycle.

Choose The Resource By Shape

Scalive gives a connected LiveView two lifecycle-owned ways to receive work later. Use ctx.async for one finite Task[A]: a database query, service call, or report that succeeds, fails, or is cancelled. Use ctx.subscriptions for a ZStream[Any, Nothing, Msg] that can emit many messages over time: a clock, notification feed, or application event stream.

A Task[A] is finite work that may fail with a Throwable; a fiber is its running, interruptible execution. A ZStream[R, E, A] emits zero or more A values over time. Scalive uses interruption to cancel work when its owner or a replacement disappears.

Use an injected service for durable or shared state. The APIs on this page own work only for one connected LiveView lifecycle.

Both APIs attach resources to the socket lifecycle. Scalive interrupts them when the socket closes, and starting replacement work cannot leak a stale completion into the current model. Prefer these APIs to fork inside a message handler: a manually forked fiber has no automatic owner, result delivery, or cleanup contract with the LiveView runtime.

Run Finite Work With A Typed Key

An AsyncKey[A] names one task and fixes its result type. Derive keys from stable instance identity when multiple copies of an example or component can coexist:

scala
private val ReportTask = AsyncKey[Report](s"async-report-$instanceId")

Start work through ctx.async.start and map its result into the owning LiveView's message type:

scala
ctx.async
  .start(ReportTask)(generateReport)(Msg.ReportCompleted(_))
  .as(model.copy(report = model.report.loading()))

The mapper receives a LiveAsyncResult[A]: Succeeded(value), Failed(cause), or Cancelled(reason). This is one task's completion, not persistent UI state. Handle it as an ordinary typed message so the model remains the single input to rendering.

Starting the same key again interrupts and replaces its previous task. The obsolete task does not emit Cancelled, and its stale completion is not delivered. This makes replacement suitable for refreshes and changing queries: start the newest request under the same key instead of assigning request IDs and filtering old results yourself.

Model Finite Work With AsyncValue

AsyncValue[A] is an optional application model for rendering work across messages. It distinguishes empty, loading, successful, failed, and explicitly cancelled states. Loading, failure, and cancellation can retain the last successful value, which lets a refresh show existing data instead of replacing the entire panel with a spinner.

scala
final case class Model(report: AsyncValue[Report] = AsyncValue.empty)

case Msg.ReportCompleted(result) =>
  ZIO.succeed(model.copy(report = model.report.updated(result)))

Call loading(reset = true) when old data would be misleading. The default retains it. Render every state deliberately; a failed task is domain-visible state, not a reason for the LiveView process to crash.

Do not expose arbitrary exception messages to visitors. Convert expected failures into stable, actionable copy and log diagnostic context separately. The report example displays a fixed failure explanation while its X-ray projector records only the state label.

Cancel Explicitly When It Is User-Visible

ctx.async.cancel interrupts active work. If the key exists, Scalive invokes the original mapper with LiveAsyncResult.Cancelled, including the optional application reason. Use this path when cancellation itself belongs in the UI:

scala
ctx.async.cancel(ReportTask, Some("Cancelled by the user")).as(model)

Cancelling an absent key is a no-op. Replacement, component removal, and socket shutdown also interrupt work, but intentionally do not deliver cancellation messages: their owner is obsolete or disappearing.

Reset needs an explicit policy. The example cancels active work and immediately returns to AsyncValue.Empty; it then ignores the cancellation completion only when the model is already empty. A reset that should display “cancelled” can instead reuse the normal cancel path.

Own Long-Lived Streams With Subscriptions

A SubscriptionKey identifies one registered stream in a root LiveView. The stream must emit the LiveView's Msg type and cannot fail:

scala
private val ClockSubscription =
  SubscriptionKey(s"subscription-clock-$instanceId")

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

start rejects a duplicate active key. Use it when starting twice indicates a state-machine mistake. The clock guards start with its model so the button and server transition agree.

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, or another stream parameter is valid application behavior.

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:

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

Subscription messages pass through info lifecycle hooks before handleMessage. Registrations exist only for the connected socket and do not run during disconnected rendering. Mount therefore starts required streams again for each new connected lifecycle.

Keep Ownership Local

Keys are runtime identities, not persistence keys. Keep them stable within one owner and unique among that owner's resources. Nested LiveViews already have independent resource namespaces, but deriving keys from instance identity also makes ownership visible in traces and prevents collisions when code moves into a shared owner.

Keep durable results in an application service or database when they must survive navigation or reconnect. AsyncValue, task registrations, and subscription registrations are socket state. On page exit, Scalive releases the managed resource; on remount, initialize the model and registrations from their 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))

  override 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-92)

The report implementation shows typed results, retained values, deterministic failure, stale-completion suppression, explicit cancellation, retry, and reset:

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

  private val ReportTask = reportKey(instanceId)

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

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.RunSuccess => start(model, ctx, successfulReport)
    case Msg.RunFailure => start(model, ctx, failingReport)
    case Msg.Replace    => start(model, ctx, replacementReport)
    case Msg.Retry      => start(model, ctx, retryReport)
    case Msg.Cancel     =>
      if model.report.isLoading then
        ctx.async.cancel(ReportTask, Some("Cancelled by the user")).as(model)
      else ZIO.succeed(model)
    case Msg.Reset =>
      ctx.async.cancel(ReportTask, Some("Example reset")).as(Model())
    case Msg.ReportCompleted(LiveAsyncResult.Cancelled(_)) if model.report == AsyncValue.Empty =>
      ZIO.succeed(model)
    case Msg.ReportCompleted(result) =>
      ZIO.succeed(model.copy(report = model.report.updated(result)))

  override def view(model: Signal[Model]): HtmlElement[Msg] =
    val report = model.map(_.report)
    div(
      cls := "docs-managed-work",
      div(
        cls := "docs-managed-work-controls",
        button(typ := "button", on.click(Msg.RunSuccess), "Run successful report"),
        button(typ := "button", on.click(Msg.RunFailure), "Run failing report"),
        button(typ := "button", on.click(Msg.Replace), "Replace current work"),
        button(typ := "button", on.click(Msg.Retry), "Retry report"),
        button(
          typ      := "button",
          disabled := report.map(!_.isLoading),
          on.click(Msg.Cancel),
          "Cancel report"
        )
      ),
      report
        .map(_ == AsyncValue.Empty).choose(
          sectionTag(
            dataAttr("report-state") := "",
            aria.live                := "polite",
            "Empty"
          ),
          reportPanel(
            report.map(reportState),
            report.map(reportValue),
            report.map(reportStatus)
          )
        )
    )
  end view

  private def start(model: Model, ctx: MessageContext, task: Task[Report]) =
    ctx.async
      .start(ReportTask)(task)(Msg.ReportCompleted(_))
      .as(model.copy(report = model.report.loading()))

  private def reportPanel(
    state: Signal[String],
    report: Signal[Option[Report]],
    status: Signal[String]
  ): HtmlElement[Msg] =
    sectionTag(
      dataAttr("report-state") := "",
      aria.live                := "polite",
      h2(dataAttr("report-status") := "", state),
      p(status),
      report.option { value =>
        articleTag(
          h3(dataAttr("report-title") := "", value.map(_.title)),
          p(value.map(_.summary)),
          p(value.map(value => s"${value.rows} rows"))
        )
      }
    )

  private def reportState(value: AsyncValue[Report]): String = value match
    case AsyncValue.Empty           => "Empty"
    case AsyncValue.Loading(_)      => "Loading"
    case AsyncValue.Ok(_)           => "Succeeded"
    case AsyncValue.Failed(_, _)    => "Failed"
    case AsyncValue.Cancelled(_, _) => "Cancelled"

  private def reportValue(value: AsyncValue[Report]): Option[Report] = value match
    case AsyncValue.Empty                  => None
    case AsyncValue.Loading(previous)      => previous
    case AsyncValue.Ok(report)             => Some(report)
    case AsyncValue.Failed(previous, _)    => previous
    case AsyncValue.Cancelled(previous, _) => previous

  private def reportStatus(value: AsyncValue[Report]): String = value match
    case AsyncValue.Empty                => "Empty"
    case AsyncValue.Loading(_)           => "Generating report..."
    case AsyncValue.Ok(_)                => "Report completed."
    case AsyncValue.Failed(_, _)         => "The deterministic data source rejected the report."
    case AsyncValue.Cancelled(_, reason) =>
      reason.getOrElse("Report generation was cancelled.")
end AsyncReportExample

object AsyncReportExample:
  final case class Report(title: String, rows: Int, summary: String)
  final case class Model(report: AsyncValue[Report] = AsyncValue.empty)

  enum Msg:
    case RunSuccess
    case RunFailure
    case Replace
    case Retry
    case Cancel
    case Reset
    case ReportCompleted(result: LiveAsyncResult[Report])

  private[docs] def reportKey(instanceId: String): AsyncKey[Report] =
    AsyncKey[Report](s"async-report-$instanceId")

  private def successfulReport: Task[Report] =
    ZIO
      .sleep(2.seconds).as(
        Report("Quarterly activity", 128, "The deterministic success path completed normally.")
      )

  private def failingReport: Task[Report] =
    ZIO.sleep(1.second) *>
      ZIO.fail(new RuntimeException("The deterministic data source rejected the report."))

  private def replacementReport: Task[Report] =
    ZIO
      .sleep(600.millis).as(
        Report("Replacement report", 64, "The replacement suppressed the obsolete completion.")
      )

  private def retryReport: Task[Report] =
    ZIO
      .sleep(800.millis).as(
        Report("Retried report", 128, "The retry completed with deterministic data.")
      )
end AsyncReportExample

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

Run the managed clock subscription and managed async report alongside their X-ray views to correlate messages, model transitions, and final DOM changes.