first of all, a consumer is very memory intensive, so you generally don't want more than one
second, having two consumers in the same app does not really increase robustness
if the error you have is at code level, it's much easier (and faster)to catch it and recover in code than it is to go through a rebalancing cycle
if you're error isn't caught (and the service goes down), then both of your consumers will be down, so nothing is consumed either
what you want is a single consumer in your app, but deploy several instances of it
f so, how does one orchestrate the consumer streams?
you don't need to do this explicitly, once you deploy multiple services, each with one consumer, in the same consumer group, they will coordinate among themselves to divvy up the partitions of the topic
in your case, because you have only one partition, they will basically be active-passive
E.g. I can see
delete only has a test for whether it fails if it's non existant
I don't particularly have doubts, but as I was doing it I thought to myself that it's a little odd. I guess technically
Sync allows SyncIO whereas Concurrent doesn't? Maybe it's not important though. I'll leave it to yourself and other reviewers make that judgement :smile:
If you really want both Sync or Concurrent, you can take a Compiler.Target constraint instead
Then Concurrent is best
heyas, i have prior implementation for merging (sorted) two infinite lazylists, and it uses this:
def monotonicAscendingMerge[A](as: Stream[Pure, A], bs: Stream[Pure, A])(implicit ev: A => Ordered[A]): Stream[Pure, A] = {
(as, bs) match {
case (Stream.empty, _) => bs
case (_, Stream.empty) => as
case (aHead#::aTail, bHead#::bTail) =>
if (aHead < bHead) aHead #:: monotonicAscendingMerge(aTail, bs)
else bHead #:: monotonicAscendingMerge(as, bTail)
}
well, it was lazy list, but i'm converting to fs2 stream, but new to this
any suggestions on how to convert?
probably a dumb question, but is something i should be doing with pull? what is the shorthand cons pattern matching operator like #::?
i ended up building on another comment I found in here and came up with the following:
def monotonicAscendingMerge[F[_], A](as: Stream[F, A], bs: Stream[F, A])(implicit ev: A => Ordered[A]): Stream[F, A] = {
// take two streams or ordered types that are monotonically increasing and
// create a result stream that merges the input streams to
// produce an output that is monotonically increasing
def go(inL: Option[(A, Stream[F, A])], inR: Option[(A, Stream[F, A])]): Pull[F, A, Unit] =
(inL, inR) match {
case (None, None) => Pull.done
case (Some((lhs, lhsNext)), None) =>
Pull.output1(lhs) >> lhsNext.pull.uncons1.flatMap(go(_, None))
case (None, Some((rhs, rhsNext))) =>
Pull.output1(rhs) >> rhsNext.pull.uncons1.flatMap(go(None, _))
case (Some((lhs, lhsNext)), Some((rhs, rhsNext))) =>
if (lhs < rhs)
Pull.output1(lhs) >> lhsNext.pull.uncons1.flatMap(go(_, inR))
else
Pull.output1(rhs) >> rhsNext.pull.uncons1.flatMap(go(inL, _))
}
as.pull.uncons1.flatMap(uca => bs.pull.uncons1.flatMap(ucb => go(uca, ucb))).stream
}
it seems better further because it handles the case where a stream is empty now, but in a future state it may not be empty?
you need
StepLeg when you Pull from more than one stream
(which is kind of an unfortunate gotcha, but still)
i will ask, what is the danger using uncons across multiple streams vs stepleg? what can happen?
thank you!
hi not sure if this is a fs2 question or a http4s question:
I’ve got an async api that returns a Stream of Rows :
I’ve got an async api that returns a Stream of Rows :
def data(id: Int): Stream[F, Row] = {
for {
q <- Stream.eval(Queue.noneTerminated[F, Row])
_ <- Stream.eval { Concurrent[F].delay {
someApi.asyncData(id, new AsyncDataHandler {
override def onRow(r: Row): Unit = q.enqueue1(Some(r))
override def onEnd(): Unit = q.enqueue1(None)
})
}
}
r <- q.dequeue
} yield r
}
in http4s, i have:
val rowsEndpoint = HttpRoutes.of[IO] {
case GET -> Root / “data" / IntVar(id) => {
Ok(data(id).map(r =>
s""”{"col1": ${r.col1()}}"""
))
}
}
i don’t seem to be getting anything back from the http4s server when hitting the endpoint
i’ve debugged, and the
onRow() and onEnd() are being hit
however it seems maybe the Stream never terminates???