Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced
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
Hi all. New to fs2 and having a question. Why fs2.Stream does not implement
Traverse like other Collection classes? Say I have the same situation in https://stackoverflow.com/questions/53806418/chaining-a-number-of-transitions-with-the-state-monad
But the question is, fs2.Stream has no
traverse, so the answer does not work in my situation
You'll notice that most effect types lack such an instance, any
IO monad can't because it represents a non-referentially transparent computation
Stream has /analogous/ operations to traverse, like
evalMap and others, but the "leave stream" part needs to go through the .compile step
It's possible that
Stream for F=Pure could have such an instance
but any general
F couldn't
