Loading data and live updates

Define a service

So far every page has kept its data in its own model. Real pages read data from databases, other services, or devices. This lesson builds a thermostat page that reads a temperature sensor when the page connects and refreshes the reading every ten seconds.

Put the sensor behind a service. Create app/src/quickstart/Thermostat.scala:

package quickstart

import zio.*

trait Thermostat:
  def temperature: Task[Double]

object Thermostat:
  /** Stands in for a real sensor, which takes a moment to answer. */
  val simulated: ULayer[Thermostat] = ZLayer.succeed(
    new Thermostat:
      def temperature: Task[Double] =
        Random.nextDoubleBetween(19.0, 23.0).delay(500.millis)
  )

Thermostat is an ordinary Scala trait, so a test can replace it with a predictable implementation. The ZLayer describes how the application builds the real one at startup. This stand-in returns a random temperature after half a second, like a slow device would.

Give the page its service

Add a route to Routes.scala:

val thermostat = live / "thermostat"

In Main.scala, register the route with from instead of ->. It receives the service and builds the LiveView:

Routes.thermostat.from((thermostat: Thermostat) =>
  ThermostatLiveView(thermostat)
),

The route now requires a Thermostat, and so do the routes built from the application. Provide the layer where the server starts:

_ <- Server.serve(routes).provide(Server.defaultWithPort(port), Thermostat.simulated)

If Main does not provide it, the program does not compile. Scalive calls the from function for each mount, so the LiveView can be a plain class that takes the service in its constructor. The services guide covers several services, layers that need resources, and routes with parameters.

Load data when the page connects

A LiveView mounts twice: once to render the first HTML response, and again when the browser connects. Reading a slow sensor during the first mount would delay the page, and its result would be discarded when the connected mount builds a new model. Read it in the connected mount instead:

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)

ctx.connection is Connection.Disconnected during the first render and Connection.Connected(connected) once the browser has connected. Only the connected branch has connected.async and connected.subscriptions, so the compiler prevents starting background work during the first render. Both branches return a model whose reading is AsyncValue.loading.

connected.async.start runs the Task in the background and returns immediately, so the connected page renders without waiting. The key and the result-to-message function come from the companion object:

final case class Model(reading: AsyncValue[Double])

enum Msg:
  case Refresh
  case Measured(result: LiveAsyncResult[Double])

private val Reading      = AsyncKey[Double]("reading")
private val RefreshTimer = SubscriptionKey("refresh-timer")
private val RefreshEvery = 10.seconds

AsyncKey[Double] names the task and fixes its result type. When the task finishes, Scalive sends Msg.Measured(result), where result is a LiveAsyncResult[Double]: Succeeded(value), Failed(cause), or Cancelled(reason). The task stops if the visitor leaves the page.

Refresh on a timer

The second statement in the connected branch starts a subscription. connected.subscriptions.start runs a ZStream for as long as the page stays connected and sends each value it emits as a message. This stream emits Msg.Refresh every ten seconds. SubscriptionDelivery.Latest skips refreshes that queue up while the page is busy, because only the newest one matters.

Msg.Refresh starts the same task again, and the Refresh button sends the same message:

def handleMessage(model: Model, ctx: MessageContext) =
  case Msg.Refresh =>
    ctx.async
      .start(Reading)(thermostat.temperature)(Msg.Measured(_))
      .as(model.copy(reading = model.reading.loading()))
  case Msg.Measured(result) =>
    ZIO.succeed(model.copy(reading = model.reading.updated(result)))

Starting a key that is already running replaces the old task, so a slow reading can never overwrite a newer one. loading() marks the reading as loading while keeping the last value, and updated(result) applies the finished result.

Render each state

The view renders every state of the reading:

def view(model: Signal[Model]): HtmlElement[Msg] =
  val reading = model.map(_.reading)
  mainTag(
    h1("Thermostat"),
    p("Temperature: ", outputTag(reading.map(_.valueOption.fold("unknown")(celsius)))),
    reading.map(_.isLoading).when(p("Reading the sensor…")),
    reading.map(failed).when(p(role := "alert", "The sensor did not answer.")),
    button(on.click(Msg.Refresh), "Refresh")
  )

valueOption is the last successful reading, so the temperature stays visible while a refresh runs or after it fails. isLoading shows the progress message, and the alert appears only after a failure. The visitor sees a fixed message rather than the exception.

Try it

The complete app/src/quickstart/ThermostatLiveView.scala:

package quickstart

import zio.*
import zio.stream.ZStream

import scalive.*

final class ThermostatLiveView(thermostat: Thermostat)
    extends LiveView[ThermostatLiveView.Msg, ThermostatLiveView.Model]:
  import ThermostatLiveView.*

  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)

  def handleMessage(model: Model, ctx: MessageContext) =
    case Msg.Refresh =>
      ctx.async
        .start(Reading)(thermostat.temperature)(Msg.Measured(_))
        .as(model.copy(reading = model.reading.loading()))
    case Msg.Measured(result) =>
      ZIO.succeed(model.copy(reading = model.reading.updated(result)))

  def view(model: Signal[Model]): HtmlElement[Msg] =
    val reading = model.map(_.reading)
    mainTag(
      h1("Thermostat"),
      p("Temperature: ", outputTag(reading.map(_.valueOption.fold("unknown")(celsius)))),
      reading.map(_.isLoading).when(p("Reading the sensor…")),
      reading.map(failed).when(p(role := "alert", "The sensor did not answer.")),
      button(on.click(Msg.Refresh), "Refresh")
    )
end ThermostatLiveView

object ThermostatLiveView:
  final case class Model(reading: AsyncValue[Double])

  enum Msg:
    case Refresh
    case Measured(result: LiveAsyncResult[Double])

  private val Reading      = AsyncKey[Double]("reading")
  private val RefreshTimer = SubscriptionKey("refresh-timer")
  private val RefreshEvery = 10.seconds

  private def celsius(value: Double): String = f"$value%.1f °C"

  private def failed(reading: AsyncValue[Double]): Boolean = reading match
    case AsyncValue.Failed(_, _) => true
    case _                       => false

Run mill app.run and open http://localhost:8080/thermostat. The page shows Reading the sensor…, then a temperature about half a second later. The reading changes every ten seconds, and Refresh reads it immediately. View the page source: the first HTML response contains Reading the sensor…, because the sensor is only read after the browser connects.

Choose where to load

Load in the connected branch when the data is slow or only matters to visitors with a live connection. Load directly in mount when the data is fast and should be in the first HTML response, for example for search engines or visitors whose JavaScript has not loaded yet. Both mounts then run the query.

When a service can announce changes itself, subscribe to its stream rather than polling. The stream maps each change to a message:

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

The asynchronous work and subscriptions guides cover cancellation, replacing a stream, and work owned by components. Next, URL state and navigation keeps page state in the address bar.