Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced
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
btw with Stream you can have stateful computations that don't require the use of
StateT, but there is nothing wrong with doing what you're doing at first, so definitely keep exploring :)
genuine question
@Daenyth in fairness, the same can be said of
compile.list :P
but it's at least used in a monomorphic context
the traverse instance means that code could be calling traverse not knowing F=Stream and assuming finite-ness
and it kind of depends on your philosophical stance whether returning
Nothing for a method impl satisfies the contract or not
Well just like this piece of code, https://scastie.scala-lang.org/Nuz2i83yTnWs0IPtOUhPBg, I just have no idea what the
evalMap is... Any advice?
takes an element, wraps it in
F[_], then evaluates the F to get the element back
All your StateT are using
pure so it might be easier to skip the T and just use State
if you just have three stateT computations, you can
Stream.eval(first) ++ Stream.eval(second) ++ Stream.eval(third) but then why are you using Stream? Could just have first >> second >> third