ascent

Datastar HTTP

ascent-datastar-http makes a heddle server "an ascent client": render a UI with ascent-html, push patch-elements / patch-signals through heddle's Datastar SSE generator.

patchSignal

Handler-level assert: the SSE body carries a datastar-patch-signals frame with the named value.

{
  val sse = events {
    AscentDatastar.patchSignal("count", 42)
  }
  (for
    response <- sse.run(Request.get("/sse"))
    body     <- response.body.utf8
  yield (
    response.status == Status.Ok,
    body.contains("event: datastar-patch-signals"),
    body.contains("\"count\":42"),
  )).orDie
}
(true,true,true)

Live counter server

End-to-end against an in-process server (ephemeral port, scoped to the example): POST /increment succeeds while the server is live, then the Scope closes and the port is released.

{
  case class State(count: Ref[Int], pulse: Hub[Unit])

  def pushCount(state: State): ZIO[Datastar, Nothing, Unit] =
    state.count.get.flatMap(c => AscentDatastar.patchSignal("count", c)(using JsonEncoder.int))

  def routes(state: State): Routes[Any, Nothing] =
    Routes(
      Method.GET / "sse" -> events {
        for
          _      <- pushCount(state)
          stream <- state.pulse.subscribe.map(zio.stream.ZStream.fromQueue(_))
          _      <- stream.mapZIO(_ => pushCount(state)).runDrain
        yield ()
      },
      Method.POST / "increment" -> handler { (_: Request) =>
        (state.count.update(_ + 1) *> state.pulse.publish(()).unit).as(Response.ok)
      },
    )

  for
    count <- Ref.make(0)
    pulse <- Hub.unbounded[Unit]
    state = State(count, pulse)
    result <- ZIO
      .scoped {
        Server.install(routes(state)).flatMap { server =>
          server.port.flatMap { port =>
            Client.request(Method.POST, s"http://127.0.0.1:$port/increment").map(_.status)
          }
        }
      }
      .provide(Server.defaultWith(_.port(Port(0))))
  yield result
  end for
}
Status(200,OK)