Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced
normal
zip processes one stream and then the other
also @disblader , starting with
zip like stuff to learn Pull is a unfortunate because you need to be aware of stepLeg. I'd recommend using Pull on only one stream to get familiar with it, it's actually pretty simple
I stumbled across this issue today trying to build my service as a graalvm native image -- functional-streams-for-scala/fs2#1237 has anybody come up with a way to get around it?
Or have a better alternative?
evalOn means that it operates on a thread pool dedicated for thread-blocking operations
which allows callers to run them in parallel
well, normally you'd use evalOn with
Blocker and not any old EC
I recommend using it, it will make the upgrade easier
but c-e is an easy upgrade for the most part
anyway -
evalOn only changes what EC the IO runs on
it doesn't make it return immediately. Just that when it runs, it runs on that EC instead of the EC backing the ContextShift
This allows you to parallelize them much higher - suppose it's waiting on network calls but needs little cpu. You could do eg 64 of them in parallel even if you only have 8 cores, whereas if the thread-blocking call wasn't shifted with
evalOn, it would fully consume your 8 cores to wait for the blocking call
concurrency is basically always explicit with c-e/fs2
what types are you working with?
basically
evalMap(...).foldMonoid or evalMap(...).compile.foldMonoid should give you what you need in the two cases
like, either you do recursive
Pull
or, a non-concurrent stateful operator (start with
mapAccumulate, then see if you need more powerful)
or, if all fails, concurrency and/or
Ref (sometimes Ref is easier for non-concurrent stuff as well, at which point it becomes a matter of style/preference)
what are trying to do in terms of logic?
are three mutually exclusive approaches
as a guideline ofc, I'm sure there are plenty of cases where this statement is false
although I do have a vague memory that we have an op that does something similar somewhere
@milanvdm a quick-and-dirty solution using
It's by far not optimized (String concatenation and similar), but it should give an idea on how it could be done.
mapAccumulate: https://scastie.scala-lang.org/CBzo7dS0QGCzvTnbwUGJGQIt's by far not optimized (String concatenation and similar), but it should give an idea on how it could be done.
btw I was thinking of
repartition. but I'm not sure it fits your case https://github.com/functional-streams-for-scala/fs2/blob/master/core/shared/src/main/scala/fs2/Stream.scala#L2373
I have a use case where streamA publishes to some internal (bounded) queue while streamB consumes from said queue and publishes data to some external system. My current approach is something like this:
trait MyPublisher[A] {
def publish(in: A): F[Unit]
}
// publisher should complete external publishing of all elements
// internally published before the stream terminates
def makeSafePublisher[A]: Stream[F, MyPublisher] {
val queue = Queue.boundedNoneTerminated[F, A](maxSize)
val internalPublisher = new MyPublisher[A] {
override def publish(in: A) = queue.enqueue1(in)
}
val internalPublish = Stream.emit(internalPublisher).onComplete(queue.enqueue1(None))
val externalPublish = queue.dequeue.evalMap(publishToKafka)
internalPublish concurrently externalPublishHowever, it seems the system can shutdown while elements remain in the internal queue. What is the safest way to implement makeSafePublisher that will ensure the internal queue is fully drained?
Or, reasonable alternative approach
hm, that reminds me of this conversation the other day: https://gitter.im/functional-streams-for-scala/fs2?at=5e849fd48bb4966bc49bf03b