phfroidmont/scalive
1
0
Fork 0
mirror of https://github.com/phfroidmont/scalive.git synced 2025-12-25 05:26:59 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix multiple routes support

Browse source
This commit is contained in:
Paul-Henri Froidmont 2025-09-12 17:35:11 +02:00
parent 44ffa55cc6
commit a036e3cbb3
Signed by: phfroidmont
GPG key ID: BE948AFD7E7873BE
6 changed files with 217 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

2
build.mill
View file

@ -38,7 +38,7 @@ object zio extends ScalaCommon:
object example extends ScalaCommon: object example extends ScalaCommon:
def moduleDeps = Seq(zio) def moduleDeps = Seq(zio)
def mvnDeps = Seq(mvn"dev.optics::monocle-core:3.1.0") def mvnDeps = Seq(mvn"dev.optics::monocle-core:3.1.0", mvn"dev.zio::zio-logging:2.5.1")
def scaliveBundle = Task { def scaliveBundle = Task {
os.copy( os.copy(

51
example/src/ExampleLiveView.scala → example/src/CounterLiveView.scala
View file

@ -1,27 +1,20 @@
import ExampleLiveView.* import CounterLiveView.*
import monocle.syntax.all.* import monocle.syntax.all.*
import scalive.* import scalive.*
import zio.* import zio.*
import zio.json.* import zio.json.*
import zio.stream.ZStream import zio.stream.ZStream
class ExampleLiveView(someParam: String) extends LiveView[Msg, Model]: class CounterLiveView() extends LiveView[Msg, Model]:
def init = ZIO.succeed( def init = ZIO.succeed(
Model( Model(
isVisible = true, isVisible = true,
counter = 0, counter = 0
elems = List(
NestedModel("a", 10),
NestedModel("b", 15),
NestedModel("c", 20)
)
) )
) )
def update(model: Model) = def update(model: Model) =
case Msg.IncAge(value) =>
ZIO.succeed(model.focus(_.elems.index(2).age).modify(_ + value))
case Msg.ToggleCounter => case Msg.ToggleCounter =>
ZIO.succeed(model.focus(_.isVisible).modify(!_)) ZIO.succeed(model.focus(_.isVisible).modify(!_))
case Msg.IncCounter => case Msg.IncCounter =>
@ -31,38 +24,13 @@ class ExampleLiveView(someParam: String) extends LiveView[Msg, Model]:
def view(model: Dyn[Model]) = def view(model: Dyn[Model]) =
div( div(
cls := "max-w-2xl mx-auto bg-white shadow rounded-2xl p-6 space-y-6",
h1( h1(
cls := "text-2xl font-semibold tracking-tight text-gray-900", cls := "text-2xl font-semibold tracking-tight text-gray-900",
someParam "Counter with auto increment every second"
),
cls := "max-w-2xl mx-auto bg-white shadow rounded-2xl p-6 space-y-6",
idAttr := "42",
ul(
cls := "divide-y divide-gray-200",
model(_.elems).splitByIndex((_, elem) =>
li(
cls := "py-3 flex flex-wrap items-center justify-between gap-2",
span(
cls := "text-gray-700",
"Nom: ",
span(cls := "font-medium", elem(_.name))
),
span(
cls := "text-sm text-gray-500",
"Age: ",
span(cls := "font-semibold text-gray-700", elem(_.age.toString))
)
)
)
), ),
div( div(
cls := "flex flex-wrap items-center gap-3", cls := "flex flex-wrap items-center gap-3",
button(
cls := "inline-flex items-center rounded-lg px-3 py-2 text-sm font-medium bg-gray-900 text-white shadow hover:opacity-90 focus:outline-none focus:ring-2 focus:ring-gray-900/30",
phx.click := Msg.IncAge(1),
"Inc age"
),
span(cls := "grow"),
button( button(
cls := "inline-flex items-center rounded-lg px-3 py-2 text-sm font-medium ring-1 ring-inset ring-gray-300 text-gray-700 hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-gray-400/30", cls := "inline-flex items-center rounded-lg px-3 py-2 text-sm font-medium ring-1 ring-inset ring-gray-300 text-gray-700 hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-gray-400/30",
phx.click := Msg.ToggleCounter, phx.click := Msg.ToggleCounter,
@ -98,18 +66,15 @@ class ExampleLiveView(someParam: String) extends LiveView[Msg, Model]:
def subscriptions(model: Model) = def subscriptions(model: Model) =
ZStream.tick(1.second).map(_ => Msg.IncCounter).drop(1) ZStream.tick(1.second).map(_ => Msg.IncCounter).drop(1)
end ExampleLiveView end CounterLiveView
object ExampleLiveView: object CounterLiveView:
enum Msg derives JsonCodec: enum Msg derives JsonCodec:
case IncAge(value: Int)
case ToggleCounter case ToggleCounter
case IncCounter case IncCounter
case DecCounter case DecCounter
final case class Model( final case class Model(
isVisible: Boolean, isVisible: Boolean,
counter: Int, counter: Int)
elems: List[NestedModel])
final case class NestedModel(name: String, age: Int)

28
example/src/Example.scala
View file

@ -1,18 +1,39 @@
import scalive.{label as _, *}
import zio.* import zio.*
import zio.http.* import zio.http.*
import scalive.* import zio.logging.ConsoleLoggerConfig
import zio.logging.LogColor
import zio.logging.LogFilter
import zio.logging.LogFormat.*
import zio.logging.consoleLogger
object Example extends ZIOAppDefault: object Example extends ZIOAppDefault:
private val logFormat =
label("timestamp", timestamp.fixed(32)).color(LogColor.BLUE) |-|
label("level", level.fixed(5)).highlight |-|
label("thread", fiberId).color(LogColor.WHITE) |-|
label("message", quoted(line)).highlight |-|
cause
val logFilter = LogFilter.LogLevelByNameConfig(LogLevel.Debug)
override val bootstrap =
Runtime.removeDefaultLoggers >>> consoleLogger(ConsoleLoggerConfig(logFormat, logFilter))
val liveRouter = val liveRouter =
LiveRouter( LiveRouter(
RootLayout(_), RootLayout(_),
List( List(
LiveRoute( LiveRoute(
Root, Root / "counter",
(_, _) => CounterLiveView()
),
LiveRoute(
Root / "list",
(_, req) => (_, req) =>
val q = req.queryParam("q").map("Param : " ++ _).getOrElse("No param") val q = req.queryParam("q").map("Param : " ++ _).getOrElse("No param")
ExampleLiveView(q) ListLiveView(q)
) )
) )
) )
@ -20,3 +41,4 @@ object Example extends ZIOAppDefault:
val routes = liveRouter.routes @@ Middleware.serveResources(Path.empty / "static", "public") val routes = liveRouter.routes @@ Middleware.serveResources(Path.empty / "static", "public")
override val run = Server.serve(routes).provide(Server.default) override val run = Server.serve(routes).provide(Server.default)
end Example

72
example/src/ListLiveView.scala Normal file
View file

@ -0,0 +1,72 @@
import ListLiveView.*
import monocle.syntax.all.*
import scalive.*
import zio.*
import zio.json.*
import zio.stream.ZStream
class ListLiveView(someParam: String) extends LiveView[Msg, Model]:
def init = ZIO.succeed(
Model(
elems = List(
NestedModel("a", 10),
NestedModel("b", 15),
NestedModel("c", 20)
)
)
)
def update(model: Model) =
case Msg.IncAge(value) =>
ZIO.succeed(model.focus(_.elems.index(2).age).modify(_ + value))
def view(model: Dyn[Model]) =
div(
h1(
cls := "text-2xl font-semibold tracking-tight text-gray-900",
someParam
),
cls := "max-w-2xl mx-auto bg-white shadow rounded-2xl p-6 space-y-6",
idAttr := "42",
ul(
cls := "divide-y divide-gray-200",
model(_.elems).splitByIndex((_, elem) =>
li(
cls := "py-3 flex flex-wrap items-center justify-between gap-2",
span(
cls := "text-gray-700",
"Nom: ",
span(cls := "font-medium", elem(_.name))
),
span(
cls := "text-sm text-gray-500",
"Age: ",
span(cls := "font-semibold text-gray-700", elem(_.age.toString))
)
)
)
),
div(
cls := "flex flex-wrap items-center gap-3",
button(
cls := "inline-flex items-center rounded-lg px-3 py-2 text-sm font-medium bg-gray-900 text-white shadow hover:opacity-90 focus:outline-none focus:ring-2 focus:ring-gray-900/30",
phx.click := Msg.IncAge(1),
"Inc age"
),
span(cls := "grow")
)
)
def subscriptions(model: Model) = ZStream.empty
end ListLiveView
object ListLiveView:
enum Msg derives JsonCodec:
case IncAge(value: Int)
final case class Model(
elems: List[NestedModel])
final case class NestedModel(name: String, age: Int)

120
zio/src/scalive/LiveRouter.scala
View file

@ -61,19 +61,19 @@ class LiveChannel(private val sockets: SubscriptionRef[Map[String, Socket[?, ?]]
lv: LiveView[Msg, Model], lv: LiveView[Msg, Model],
meta: WebSocketMessage.Meta meta: WebSocketMessage.Meta
): RIO[Scope, Unit] = ): RIO[Scope, Unit] =
sockets.updateZIO { m => sockets
m.get(id) match .updateZIO { m =>
case Some(socket) => m.get(id) match
socket.shutdown *> case Some(socket) =>
socket.shutdown *>
Socket
.start(id, token, lv, meta)
.map(m.updated(id, _))
case None =>
Socket Socket
.start(id, token, lv, meta) .start(id, token, lv, meta)
.map(m.updated(id, _)) .map(m.updated(id, _))
case None => }.flatMap(_ => ZIO.logDebug(s"LiveView joined $id"))
Socket
.start(id, token, lv, meta)
.map(m.updated(id, _))
}
def event(id: String, value: String, meta: WebSocketMessage.Meta): UIO[Unit] = def event(id: String, value: String, meta: WebSocketMessage.Meta): UIO[Unit] =
sockets.get.flatMap { m => sockets.get.flatMap { m =>
@ -99,45 +99,47 @@ class LiveRouter(rootLayout: HtmlElement => HtmlElement, liveRoutes: List[LiveRo
private val socketApp: WebSocketApp[Any] = private val socketApp: WebSocketApp[Any] =
Handler.webSocket { channel => Handler.webSocket { channel =>
ZIO.scoped(for ZIO
liveChannel <- LiveChannel.make() .scoped(for
_ <- liveChannel.diffsStream liveChannel <- LiveChannel.make()
.runForeach((payload, meta) => _ <- liveChannel.diffsStream
channel .runForeach((payload, meta) =>
.send( channel
Read( .send(
WebSocketFrame.text( Read(
WebSocketMessage( WebSocketFrame.text(
meta.joinRef, WebSocketMessage(
meta.messageRef, meta.joinRef,
meta.topic, meta.messageRef,
payload match meta.topic,
case Payload.Diff(_) => "diff" payload match
case _ => "phx_reply", case Payload.Diff(_) => "diff"
payload case _ => "phx_reply",
).toJson payload
).toJson
)
) )
) )
) )
) .tapErrorCause(c => ZIO.logErrorCause("diffsStream pipeline failed", c))
.tapErrorCause(c => ZIO.logErrorCause("diffsStream pipeline failed", c)) .ensuring(ZIO.logWarning("WS out fiber terminated"))
.ensuring(ZIO.logWarning("WS out fiber terminated")) .fork
.fork _ <- channel
_ <- channel .receiveAll {
.receiveAll { case Read(WebSocketFrame.Close) => ZIO.logDebug("WS connection closed by client")
case Read(WebSocketFrame.Text(content)) => case Read(WebSocketFrame.Text(content)) =>
for for
message <- ZIO message <- ZIO
.fromEither(content.fromJson[WebSocketMessage]) .fromEither(content.fromJson[WebSocketMessage])
.mapError(new IllegalArgumentException(_)) .mapError(new IllegalArgumentException(_))
reply <- handleMessage(message, liveChannel) reply <- handleMessage(message, liveChannel)
_ <- reply match _ <- reply match
case Some(r) => channel.send(Read(WebSocketFrame.text(r.toJson))) case Some(r) => channel.send(Read(WebSocketFrame.text(r.toJson)))
case None => ZIO.unit case None => ZIO.unit
yield () yield ()
case _ => ZIO.unit case _ => ZIO.unit
}.tapErrorCause(ZIO.logErrorCause(_)) }
yield ()) yield ()).tapErrorCause(ZIO.logErrorCause(_))
} }
@ -160,15 +162,21 @@ class LiveRouter(rootLayout: HtmlElement => HtmlElement, liveRoutes: List[LiveRo
ZIO ZIO
.fromEither(URL.decode(url)).flatMap(url => .fromEither(URL.decode(url)).flatMap(url =>
val req = Request(url = url) val req = Request(url = url)
liveRoutes liveRoutes.iterator
.collectFirst { route => .map(route =>
val pathParams = route.path.decode(req.path).getOrElse(???) route.path
val lv = route.liveviewBuilder(pathParams, req) .decode(req.path)
liveChannel .toOption
.join(message.topic, session, lv, message.meta)(using route.messageCodec) .map(route.liveviewBuilder(_, req))
.map(_ => None) .map(
ZIO.logDebug(s"Joining live view ${route.path.toString} ${message.topic}") *>
}.getOrElse(ZIO.succeed(None)) liveChannel.join(message.topic, session, _, message.meta)(
using route.messageCodec
)
)
)
.collectFirst { case Some(join) => join.map(_ => None) }
.getOrElse(ZIO.succeed(None))
) )
case Payload.Event(_, event, _) => case Payload.Event(_, event, _) =>
liveChannel liveChannel

92
zio/src/scalive/Socket.scala
View file

@ -23,55 +23,57 @@ object Socket:
lv: LiveView[Msg, Model], lv: LiveView[Msg, Model],
meta: WebSocketMessage.Meta meta: WebSocketMessage.Meta
): RIO[Scope, Socket[Msg, Model]] = ): RIO[Scope, Socket[Msg, Model]] =
for ZIO.logAnnotate("lv", id) {
inbox <- Queue.bounded[(Msg, WebSocketMessage.Meta)](4) for
outHub <- Hub.unbounded[(Payload, WebSocketMessage.Meta)] inbox <- Queue.bounded[(Msg, WebSocketMessage.Meta)](4)
outHub <- Hub.unbounded[(Payload, WebSocketMessage.Meta)]
initModel <- lv.init initModel <- lv.init
modelVar = Var(initModel) modelVar = Var(initModel)
el = lv.view(modelVar) el = lv.view(modelVar)
ref <- Ref.make((modelVar, el)) ref <- Ref.make((modelVar, el))
initDiff = el.diff(trackUpdates = false) initDiff = el.diff(trackUpdates = false)
lvStreamRef <- SubscriptionRef.make(lv.subscriptions(initModel)) lvStreamRef <- SubscriptionRef.make(lv.subscriptions(initModel))
clientMsgStream = ZStream.fromQueue(inbox) clientMsgStream = ZStream.fromQueue(inbox)
serverMsgStream = (ZStream.fromZIO(lvStreamRef.get) ++ lvStreamRef.changes) serverMsgStream = (ZStream.fromZIO(lvStreamRef.get) ++ lvStreamRef.changes)
.flatMapParSwitch(1, 1)(identity) .flatMapParSwitch(1, 1)(identity)
.map(_ -> meta.copy(messageRef = None, eventType = "diff")) .map(_ -> meta.copy(messageRef = None, eventType = "diff"))
clientFiber <- clientMsgStream.runForeach { (msg, meta) => clientFiber <- clientMsgStream.runForeach { (msg, meta) =>
for for
(modelVar, el) <- ref.get (modelVar, el) <- ref.get
updatedModel <- lv.update(modelVar.currentValue)(msg) updatedModel <- lv.update(modelVar.currentValue)(msg)
_ = modelVar.set(updatedModel) _ = modelVar.set(updatedModel)
_ <- lvStreamRef.set(lv.subscriptions(updatedModel)) _ <- lvStreamRef.set(lv.subscriptions(updatedModel))
diff = el.diff() diff = el.diff()
payload = Payload.okReply(LiveResponse.Diff(diff)) payload = Payload.okReply(LiveResponse.Diff(diff))
_ <- outHub.publish(payload -> meta) _ <- outHub.publish(payload -> meta)
yield () yield ()
}.fork }.fork
serverFiber <- serverMsgStream.runForeach { (msg, meta) => serverFiber <- serverMsgStream.runForeach { (msg, meta) =>
for for
(modelVar, el) <- ref.get (modelVar, el) <- ref.get
updatedModel <- lv.update(modelVar.currentValue)(msg) updatedModel <- lv.update(modelVar.currentValue)(msg)
_ = modelVar.set(updatedModel) _ = modelVar.set(updatedModel)
diff = el.diff() diff = el.diff()
payload = Payload.Diff(diff) payload = Payload.Diff(diff)
_ <- outHub.publish(payload -> meta) _ <- outHub.publish(payload -> meta)
yield () yield ()
}.fork }.fork
stop = stop =
inbox.shutdown *> outHub.shutdown *> clientFiber.interrupt.unit *> serverFiber.interrupt.unit inbox.shutdown *> outHub.shutdown *> clientFiber.interrupt.unit *> serverFiber.interrupt.unit
outbox = outbox =
ZStream.succeed( ZStream.succeed(
Payload.okReply(LiveResponse.InitDiff(initDiff)) -> meta Payload.okReply(LiveResponse.InitDiff(initDiff)) -> meta
) ++ ZStream.unwrapScoped(ZStream.fromHubScoped(outHub)).filterNot { ) ++ ZStream
case (Payload.Diff(diff), _) => diff.isEmpty .unwrapScoped(ZStream.fromHubScoped(outHub)).filterNot {
case _ => false case (Payload.Diff(diff), _) => diff.isEmpty
} case _ => false
yield Socket[Msg, Model](id, token, inbox, outbox, stop) }
end for yield Socket[Msg, Model](id, token, inbox, outbox, stop)
}
end start end start
end Socket end Socket