fs2
fs2 copied to clipboard
typelevel
Compositional, streaming I/O library for Scala
Reproduction ([scastie](https://scastie.scala-lang.org/8IetFIeDSMSei1ldXUmmXA)): ```scala import cats.effect.std.Queue import cats.effect.IO import cats.effect.testkit.TestControl import cats.effect.unsafe.implicits.global import fs2.Stream import cats.syntax.all._ import scala.concurrent.duration._ var parallelism = 1 val queue = Queue.unbounded[IO, Int].unsafeRunSync() val drainQueue = queue.tryTakeN(None).flatMap(_.traverse_(i...
```scala Stream(1,2).covary[IO].interruptAfter(200.millis).evalMap( i => IO.println(i).delayBy(300.millis).onCancel(IO.println("CANCELED")) ).compile.drain.unsafeRunSync() ``` This program prints `CANCELED` even though this is happening downstream from `interruptAfter`. Only canceling upstream would give more control to the clients of...
This PR adds support for [mergePreferred](https://doc.akka.io/docs/akka/current/stream/operators/Source-or-Flow/mergePreferred.html) like akka provides. Internally two bounded CE Queues are used, mainly because they provide `tryTake` for peeking the queue of the preferred stream. If...
Using the latest fs2 version 3.7.0 and an effect type that includes `cats.data.EitherT`, an error in the effect is not preserved when using `fs2.io.readOutputStream`. Possibly related to #2821 Minimized reproduction:...
Tested and reproduced on version 3.2.7 and 3.6.1 This happens because in `parEvalMapUnordered`, the results of the evalMap action are passed back to the controlling stream using a callback in...
# Summary Apologies in advance for the wall of text: hopefully this time it will be worth it (if not this is probably my last attempt on this for a...
See https://discord.com/channels/632277896739946517/632310980449402880/1100649913223745597
This wants to fix #3195 Consider the case: if (size < acc.size) ```scala val (out, rem) = acc.splitAt(size - 1) Pull.output(out) >> go(rem ++ hd, -1, tl) ``` The code...
There are some really efficient APIs for this: - https://man7.org/linux/man-pages/man2/sendfile.2.html - https://docs.oracle.com/javase/8/docs/api/java/nio/channels/FileChannel.html#transferTo-long-long-java.nio.channels.WritableByteChannel- - https://docs.oracle.com/javase/8/docs/api/java/nio/channels/FileChannel.html#transferFrom-java.nio.channels.ReadableByteChannel-long-long- - https://std.in/using-sendfile-with-nodejs/ I'm fairly confident we can add a default implementation to `Socket` just using the...
Follow up to: - https://github.com/typelevel/fs2/pull/3149#issuecomment-1445226409 The idea is to support various sinks/sources for streams related to a `Process`. For example, merging stderr into stdout, or reading/writing to a file.