In Scala fs2 library for functional streams: I am trying to understand the difference between flatMap, flatTap, evalMap and evalTap. They all seem to perform the same thing, which is transformatio...

I'm just starting my adventure with fs2 streams. What I want to achieve, is to read a file (a large one, this is why I use fs2), transform it and write the result to two different files (based on s...

I have an external (that is, I cannot change it) Java API which looks like this: public interface Sender { void send(Event e); } I need to implement a Sender which accepts each event, transform...

I've got something like this (it's an example from https://github.com/typelevel/fs2, with my additions, which I marked with comments): import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource} im...

It's well documented that merging with an empty fs2.Stream should produce the same fs2.Stream. Here is the quote from Scaladocs: Has the property that merge(Stream.empty, s) == s Consider the fol...

I have an infinite fs2.Stream which may encounter errors. I'd like to skip those errors with doing nothing (probably log) and keep streaming further elements. Example: //An example val stream = fs2...

I want to do something like this: def splitStream[F, A](stream: fs2.Stream[F, A], split: A => B): (Stream[F, A], Stream[F, B)) = (stream, stream.map(split) But this does not work as it "pul...

Here is a piece of code from the documentation for fs2. The function go is recursive. The question is how do we know if it is stack safe and how to reason if any function is stack safe? import fs2...

I'm want to know how to convert Scala fs2 Stream to string, from fs2 github readme example: def converter[F[_]](implicit F: Sync[F]): F[Unit] = { val path = "/Users/lorancechen/version_control_pr...

Author of Monix says in comparison of Monix with FS2 Where FS2 is better: the model of communication between producers and consumers is pull-based, sometimes making it easier to implement new ope...

I want to run N nested streams/pipes in parallel and send each element to only one of the nested streams. Balance allows me to do this but I want to route elements with the same "key" to the same n...

I am trying to compile the stream, but somehow Compiler is not in scope, what context bound is needed for bringing it in scope? import cats.Monad def compilingStream[F[_]: Monad]: F[List[Int]] = ...

I want to use a function similar to take(n: Int) but in a time dimension: consume(period: Duration. So I want a stream to terminate if a timeout occurs. I know that I could compile a stream to some...

I have event stream as follows: sealed trait Event val eventStream: fs2.Stream[IO, Event] = //... I want to group this events received within a single minute (i.e from 0 sec to 59 sec of every ...

I have a stream of unordered measurements, that I'd like to group into batches of a fixed size, so that I can persist them efficiently later: val measurements = for { id <- Seq("foo", "bar", "...

Why isn't it possible to mock a method that returns an fs2.Stream with a Mockito mock? Mockito is complaining that I am trying to return a FreeC instead of a Stream. Why is that and how can I get...

I want to parse a json file which output a collection of A. The signature of the Output is IO[List[A]] How can I convert this value to a Stream: Stream[IO, A] ? I can convert to a Stream[IO, List[...

Extending StreamApp asks you to provide the stream def. It has a requestShutdown parameter. def stream(args: List[String], requestShutdown: F[Unit]): Stream[F, ExitCode] I provide the implementa...

While trying to become familiar with FS2, I came across a nifty recursive implementation using the Scala collections' Stream, and thought I'd have a go at trying it in FS2: import fs2.{Pure, Stre...