FS2

FS2 (Functional Streams for Scala) is a library for creating functional, effectful, and concurrent streams. It is designed to handle I/O computations (such as networking and file operations) in constant memory.

Key capabilities include:

• Constant memory I/O: Processing large data streams without exhausting memory.
• Stateful transformations: Applying logic that depends on previous elements in the stream.
• Resource safety: Ensuring effects and resources are evaluated and cleaned up correctly.
• Ecosystem integration: Built on Cats Effect and serves as the streaming engine for libraries like http4s, skunk, and doobie.

FS2 provides a declarative model for arbitrary control flow through several key capabilities:

• Zipping and merging of streams: You can read from multiple sources simultaneously and combine their elements using zipping or merging combinators. The Stream algebra allows for highly flexible topologies.
• Dynamic resource allocation: Streams can allocate resources dynamically (e.g., opening files discovered during the stream processing). FS2 guarantees these resources are released upon normal termination or when exceptions occur.
• Nondeterministic and concurrent processing: Using concurrency combinators, you can build pipelines that read from multiple inputs simultaneously, handle nondeterminism, or implement queueing at various stages. This allows streams to act as lightweight, declarative threads for complex concurrent behavior.

In FS2 1.0, streams are internally represented by Chunks instead of Segments. This results in a simpler API and better performance for most use cases. All APIs that previously returned segments now return chunks.

A regular stream of values might stop emitting if the source stops providing data, which can cause time-based aggregations to freeze. To prevent this, you can convert a regular stream into a "time series"—a stream of TimeStamped[Option[A]] where a None value represents a clock tick.

Use TimeSeries.timePulled to timestamp each received value with the wall clock time it was pulled from the source. You must specify a tick period (e.g., 1.second) to ensure the stream continues to emit None values even when no input is present.

import scala.concurrent.duration._
import cats.effect.Temporal

def withReceivedBitrate[F[_]: Temporal](input: Stream[F, Byte]): Stream[F, TimeStamped[Either[Long, Option[ByteVector]]]] =
  TimeSeries.timePulled(input.chunks.map(_.toByteVector), 1.second, 1.second).through(withBitrate)

Topic implements a general publish-subscribe pattern. It allows multiple subscribers to receive elements published to the topic. You can create a topic using Topic[F, A].flatMap (or Stream.eval(Topic[F, A])), publish elements using topic.publish (as a Pipe) or topic.publish1 (for a single element), and create subscribers using topic.subscribe(bufferSize).

import cats.effect._
import cats.effect.unsafe.implicits.global
import fs2.Stream
import fs2.concurrent.Topic

Topic[IO, String].flatMap { topic =>
  val publisher = Stream.constant("1").covary[IO].through(topic.publish)
  val subscriber = topic.subscribe(10).take(4)
  subscriber.concurrently(publisher).compile.toVector
}.unsafeRunSync()

Signal (and its extension SignallingRef) can be used to communicate state between different streams, often to trigger interruptions or changes in behavior. SignallingRef[F, A] allows you to set a new value, which can then be used by other streams via interruptWhen(signal).

import cats.effect._
import cats.effect.unsafe.implicits.global
import fs2.Stream
import fs2.concurrent.SignallingRef

import scala.concurrent.duration._

SignallingRef[IO, Boolean](false).flatMap { signal =>
  val s1 = Stream.awakeEvery[IO](1.second).interruptWhen(signal)
  val s2 = Stream.sleep[IO](4.seconds) >> Stream.eval(signal.set(true))
  s1.concurrently(s2).compile.toVector
}.unsafeRunSync()

Channel implements a publish-subscribe pattern specifically useful when you have multiple publishers and a single subscriber. You can create an unbounded channel using Channel.unbounded[F, A] and interact with it via channel.send (to publish) and channel.stream (to subscribe).

import cats.effect._
import fs2.Stream
import scala.concurrent.duration._
import cats.effect.unsafe.implicits.global
import fs2.concurrent.Channel

Channel.unbounded[IO, String].flatMap { channel =>
  val pub1 = Stream.repeatEval(IO("Hello")).evalMap(channel.send).metered(1.second)
  val pub2 = Stream.repeatEval(IO("World")).evalMap(channel.send).metered(2.seconds)
  val sub = channel.stream.evalMap(IO.println)
  Stream(pub1, pub2, sub).parJoinUnbounded.interruptAfter(6.seconds).compile.drain
}.unsafeRunSync()

FS2 has simplified its transformation model by removing specific type aliases like Process1, Tee, Wye, and Channel. These are replaced by more general Pipe and Pipe2 types:

• type Pipe[F,A,B] = Stream[F,A] => Stream[F,B]
• type Pipe2[F,A,B,C] = (Stream[F,A], Stream[F,B]) => Stream[F,C]

To apply these transformations to a stream, use the .through method for single-input pipes and .through2 for two-input pipes.

Transformation Mapping:

• Process1 functionality is now covered by Pipe.
• Tee and Wye functionality is now covered by Pipe2.
• Functions previously in the process1 module have moved to the pipe module.
• tee and wye have been combined into the pipe2 module.

// Before (0.8 or earlier)
s.pipe(process1.take(10))
s.wye(s2)(wye.blah)

// After (0.9+)
s.through(pipe.take(10))
s.through2(s2)(pipe2.blah)

FS2 1.0 leverages the cats.effect.Concurrent type class to handle cancelation. Many specialized methods in the old fs2.async package (like timedDequeue1, timedGet, or cancellableDequeue1) are no longer needed.

Instead, you can compose cancelation using Concurrent[F].race or Concurrent[F].start. For example, to implement a timed get on a Deferred (formerly Promise), use p.get.timeout(duration).