EventStream[UserID] as input, so it has something like userIdStream.flatMap(fetchUser).map(user => div(user.name)). SwitchStreamStrategy here means you don't need to think about the timing of network requests, "what if request A completes after request B, resulting in incorrect / stale output at the end). SwitchStreamStrategy is not the only way to achieve this logic, it's just the simplest one that's already implemented.
Nice! :)
I might have wanted something like this at some point, but no longer remember the use case.
Here's another interesting (and somewhat related) thing that is quite tricky to implement right now, but might be useful sometimes (I have something like this in a thing that looks like FSM but with strrams): I don't know the name for it, so I'll call it "recursive stream".
Like this:
type RecStream[T] = EventStream[(T, RecStream[T])]
val initial: RecStream[T] = ???
val result: EventStream[T] = EventStream.rec[T](initial)What do you guys think?
just curious for an actual example
I have this thing:
object AirFSM {
def bind[State]($initial: EventStream[State])(
control: PartialFunction[(Option[State], State), EventStream[State]]
): Binder[ReactiveElement.Base] = ???
}There's State, which might be something like this:
sealed trait State extends Product with Serializable
object State {
case object Initial extends State
final case class Step1() extends State
final case class Step2(inputFromStep1: String) extends State
final case class Step3(inputFromStep1: String, inputFromStep2: String) extends State
}And the bind function, which accepts:
- an initial stream of state change requests;
- and a control (partial) function which will be receiving a tuple of
(previousState, nextState)
(state transition)
The control function:
- executes whatever effects needed for the state transition at hand;
- returns a new stream of state change requests.
div(
child <-- currentStateView.signal,
AirFSM.bind[State](initialStateChangeRequests) {
case (_, State.Initial) => enterInitialState
case (Some(State.Initial), state: State.Step1) => enterStep1(state)
case (Some(_: State.Step1), state: State.Step2) => enterStep2(state)
case (Some(_: State.Step2), state: State.Step3) => enterStep3(state)
// case (prev, next) => println(s"invalid state transition: $prev -> $next")
}
)(full example: https://gist.github.com/yurique/68ec570c0b6b3cfefc316796b2863748)
(AirFSM: https://gist.github.com/yurique/e4e81c28fd5c35549b63ce7a7e232cc5)
So, having a EventStream.rec from earlier would make implementing AirFSM.bind easier (I think) :)
As well as a transitions function on a stream, but that’s easy to 3rd-party:
implicit class EventStreamExt[A](val s: EventStream[A]) extends AnyVal {
def transitions: EventStream[(Option[A], A)] = {
EventStream
.merge(
EventStream.fromValue(Option.empty[A], emitOnce = true),
s.map(a => Some(a))
).combineWith(
s.drop(1)
)
}
def drop(count: Int): EventStream[A] = {
var seen = 0
s.filter(_ => seen >= count).map { event =>
seen = seen + 1
event
}
}
def take(count: Int): EventStream[A] = {
var seen = 0
s.filter(_ => seen < count).map { event =>
seen = seen + 1
event
}
}
}
3 replies
initial param can go)
transitions, as well as take and drop, is probably not the most efficient one)
6 replies
div(“some”, nbsp, “text”) or div(s”some${nbsp}text)
I'm on the hunt for a good websockets component for laminar, and in the meantime I'm building out my own: https://github.com/keynmol/laminar-components/, could appreciate some feedback on the implementation
Don't be gentle - I'm a backend engineer and Laminar is pretty much the only thing that made me interesting in doing any frontend :D so I don't know much.
(it needs some love, but I haven’t worked with websockets in a while)
it's good to see that your modelling is similar - I want to improve mine around error handling and add some tests,.
Also want to keep it flexible as I do have some projects that are not tied to circe (ujson) or not using any non-primitive protocol at all
Observer[ClientMessage] and an EventStream[ServerMessage] so it doesn't need binding logic
I haven’t approached this in a while, but I found this in my code:
implicit class SignalObjExt(val s: Signal.type) extends AnyVal {
def seq[A](signals: Seq[Signal[A]]): Signal[Seq[A]] =
if (signals.isEmpty) {
Val(Seq.empty)
} else {
signals
.drop(1).foldLeft[Signal[Seq[A]]](signals.head.map(Seq(_)))(
(acc, next) => acc.combineWith(next).map2(_ :+ _)
)
}
}(though I don’t think I gave it too much thought back then)
def seq[A](signals: Seq[Signal[A]]): Signal[Seq[A]] =
signals.foldLeft[Signal[Seq[A]]](Val(Seq.empty))(
(acc, next) => acc.combineWith(next).map2(_ :+ _)
)this should a bit more efficient:
implicit class SignalExt[A](val s: Signal[A]) extends AnyVal {
def combineWithMap[B, C](otherSignal: Signal[B])(project: (A, B) => C): Signal[C] =
new CombineSignal2[A, B, C](
parent1 = s,
parent2 = otherSignal,
combinator = CombineObservable.guardedCombinator(project)
)
}
implicit class SignalObjExt(val s: Signal.type) extends AnyVal {
def seq[A](signals: Seq[Signal[A]]): Signal[Seq[A]] =
signals.foldLeft[Signal[Seq[A]]](Val(Seq.empty))(
(acc, next) => acc.combineWithMap(next)(_ :+ _)
)
}@raquo do you think some of this deserves to live in Airstream core? (combineWithMap, maybe named properly, seems generic enough and could optimize the s1.combineWith(s2).map2(...) use case a bit)
seq — I suppose it’s possible to make it more efficient by implementing it at a lower level with it’s own SeqSignal[A], but I have no idea how, at the moment :)
@yurique I don't think I want to expose the combine combinator because it breaks the "shared execution" expectation of Airstream. When you have stream.map(project), you know that project will be called at most once per incoming event. But that's not the case for the combine combinator, it will be called on every upstream event, and there could be more of those than resulting downstream events if one of upstream observables synchronously depends on the other (e.g. diamond case)
As for combining multiple observables at once, yes, we should be able to do this without creating N observables in the process. Just haven't had the time to implement this yet. There's a ticket about that: raquo/Airstream#18
regarding the combine — I’m not sure I understand (but I’m not insisting here :) )
but wouldn’t this
def combineWithMap[B, C](otherSignal: Signal[B])(project: (A, B) => C): Signal[C] =
new CombineSignal2[A, B, C](
parent1 = s,
parent2 = otherSignal,
combinator = CombineObservable.guardedCombinator(project)
)behave exactly like this
def combineWithMap[B, C](otherSignal: Signal[B])(project: (A, B) => C): Signal[C] =
s.combineWith(otherSignal).map { case (a, b) => project(a, b) }?
though I think I’m starting getting it :)
project will be called at most once per incoming event
Here
new CombineSignal2[A, B, C](
parent1 = s,
parent2 = otherSignal,
combinator = CombineObservable.guardedCombinator((_, _))
).map { case (a, b) => project(a, b) }tuple construction might get called more than once — (_, _) — but not the project
Actually, what exactly do you use your
seqfor?
I also have my own hand-rolled version of this (much worse than Yurii's), and I use it for dynamic interface generation - we have some specification that comes with the notion of "a list of specifications" built-in, so I can build a def seqField[A]: EventStream[Seq[A]] given I already know how to build a simpler EventStream[A]