FS2 Idioms (part 1)

The semantics of terminating

How do you know when a Stream has finished? Well, Stream.noneTerminate results in a Stream where items are wrapped in a Some and a None infers termination. Stream.unNoneTerminate does the opposite in that these wrapped items are unwrapped but terminates on the first None.

The semantics of pulling

There can be alternative behaviour when pulling from a Stream.

You could be guaranteed to get a value every time and if there is nothing new, you get the old value (Signal.continuous). In each case, the element comes in a new Stream.

Alternatively, every time you pull, you could get a new Stream that represents the latest update and if there is nothing new, the Stream is empty. This behaviour can be found in Signal.discrete.

Finally, you might want to get all the updates since you last pulled but in this case, use a `Queue` instead of Signal. This is because Signal is a "pure holder of a single value"

We can convert a Stream to a Signal by calling hold with a default value or holdOption to handle the None event.

Reading the signs
 
I had a bug in my code when I called this method on fs.concurrent.Topic:

  /**
    * Signal of current active subscribers.
    */
  def subscribers: Stream[F, Int]

My code seemed to hang. In desperation, I tried >> (which is basically flatMap { _ => ... } ) and although my Stream now seemed to start processing, the evaluated effects were not what I was expecting.

The ever helpful creator of FS2, Fabio Labella, helped me:

@PhillHenry the issue is that subscribers is an infinite stream
or rather, as long as the lifetime of the topic.
Each new element of the stream is a change in the number of subscribers
so subscribers ++ anything will never execute anything [but]
subscribers >> s2 will execute s2 every time there is a change in the number of subscribers.

If you look at the description of subscribers it says to return the current number of subscribers. If it were an F[Int], that it would be something you can poll: every time you call it, it gives you the current number and returns. But because the return type is a Stream, it means that it will emit every time there is a new subscriber, and that necessarily means it has to wait when there are no changes to the subscriber number. That is to say, it doesn't terminate.

It is not possible to know from the type whether a stream is infinite or not. In my case, if I wanted the current number of subscribers to the topic then I needed to call take(1). This way, I need not worry whether the stream is finite or not.

The take-away point is that in FP, the signatures are trying to help you to understand the semantics.

On the Pull with FS2

The learning curve for FS2 is steep and comparing it to Scala's built-in streams is not helpful. Given this Scala stream to generate Fibonacci numbers:

  val f: Stream[Int] = 0  #:: 1  #:: f.zip(f.tail).map { case (x, y) => x + y }

you might think you can do this:

    val spliced = f.take(5) ++ f.drop(5)

and treat spliced as normal:

    println(spliced.take(10).mkString(", ")) // 0, 1, 1, 2, 3, 5, 8, 13, 21, 34

And you'd be correct when working with Scala streams but not for FS2 streams, that is s != s.take(n) ++ s.drop(n)

Fabio Labella @SystemFw
I mean, this is certainly not the case in general, since the RHS reevaluates s twice

That is, s.take(n) ++ s.drop(n) will return the same values as Scala's streams but evaluate different effects to what you might expect.

For example, if we had a stream:

    rangeStream(6).evalMap { x =>
      IO {
        println(s"x = $x")
        x
      }
    }

then s.take(3) ++ s.drop(3) would indeed return a Stream containing 1,2,3,4,5,6  but you'd see this printed out:

x = 1
x = 2
x = 3
x = 1
x = 2
x = 3
x = 4
x = 5
x = 6

As Fabio says "you can use Pull if you want 'take some, then give me the rest' semantics". So, this is what I did here on GitHub.

Aside: the reason I want to splice an FS2 Stream is to introduce some test assertions. Here, I want the opposite of the above, namely to evaluate the effect (my assertion in this case) but to ignore the value. Here, one can use

Stream.eval_

 
Note the underscore. However, be a bit careful here as if there is no value returned, calls like myStream.take may semantically block.

Unit testing in Functional Programming (part 1)

Mocking in FP is considered a code smell, if not a downright antipattern but you can't avoid it with clever testing techniques. You must refactor the code.

This is what I had to do with some FS2 code from here. Now, this code demonstrates FS2's abilities wonderfully but it's not great to use in production as it's hard to test. Take this code in the link:

    val stream =
      consumerStream(consumerSettings)
        .evalTap(_.subscribeTo("topic"))
        .flatMap(_.stream)
        .mapAsync(25) { committable =>
          processRecord(committable.record)
            .as(committable.offset)
        }
        .through(commitBatchWithin(500, 15.seconds))

Not only is this hard to test, it's very dependent on Kafka specific code.

We  can abstract it so it looks like this:

  def pipeline[K, C, P, R, O, T](s:              Stream[IO, K],
                                 subscribe:      K => IO[Unit],
                                 toRecords:      K => Stream[IO, C],
                                 commitRead:     C => IO[P],
                                 producerPipe:   Pipe[IO, P, R],
                                 toWriteRecords: R => O,
                                 commitWrite:    Pipe[IO, O, T]): Stream[IO, T] = 
    s.evalTap(subscribe).flatMap(toRecords).mapAsync(25)(commitRead).through(producerPipe).map(toWriteRecords).through(commitWrite)

This makes testing easier for two reasons:

1. The code has no dependency on Kafka.
2. I can submit my own functions that use FS2's in-memory streams.

Now, the test looks like this. The domain object look like this:

  case class Kafka()

  case class Record(id: Int)

  case class ProducerRecords(id: Int)

  case class CommittableOffset(id: Int)

and although their names have a superficial resemblance to Kafka classes, this is just to make things conceptually simpler. We're dealing with a much higher layer of abstraction than that.

For instance, instead have having to connect to an embedded Kafka instance, I can have:

      val records = (1 to nToRead).map(x => Record(x))
      
      val toRecords: Kafka => Stream[IO, Record] =
        _ => Stream.emits(records).covary[IO]

and pass this as the toRecords argument to the function we want to test, pipeline. It's all in memory, no ports and sockets to worry about.

[Aside: from the docs: given a pure stream, "to convert to an effectful stream, use covary"]

The takeaway point is this: if you want to use FP to test your OO code and avoid mocks, it isn't going to work. You need to do a root and branch refactoring to get the benefits of avoiding Mockito. But your code will also be more abstract and therefore more re-usable.

Miscellaneous FS2 notes

These are my first steps into the FS2 [underscore.io] world ("when you start working with fs2 is the fact everything is a Stream. By everything I mean everything, even your Queue").

Some terminology

"Pure means no effect on the world. It just makes your CPU work and consumes some power, but besides that it does not affect the world around you.
Use evalMap instead of map when you want to apply a function that has an effect like loadUserIdByName to a Stream." [FreeCodeCamp]

"Pipe[F, A, B] is just Stream[F, A] => Stream[F, B]
I believe it's just a type alias" [Chris Davenport, Gitter].

Note the difference between concurrent and synchronization in this from FS2's Gitter channel:

Oleg Pyzhcov @oleg-py
If you have two threads doing update at the same time, one will always see results of another (not the case with var, for instance, where both can see old state) - that's safe concurrent modification. However, you can't enforce ordering of those operations with Ref alone - that's lack of synchronization.

Fabio Labella @SystemFw
FS2 concurrency builds up off a few simple primitives the most basic one being a primitive to start an F asyncly

Not needing Effect is better because Effect is very very (too) powerful, being basically equivalent to the ability to unsafeRunSync it also means that we don't need to thread ExecutionContexts around (you will only find ECs in APIs that block).

Minimal implementation

You'll notice that your main class extends cats.effect.IOApp and your 'main' methods now becomes this:

  override def run(args: List[String]): IO[ExitCode] =

IOApp pulls in a lot of implicits and an execution context, making things much easier and explaining why isolated snippets might not compile.

captainmannering @captainmannering Apr 07 13:59
I would like to use Cats IO and fs2 to write a program with the following structure: one task that receives events from one data source (say RabbitMQ), another task that receives events from another data source (say Kafka) and both of these are then sent into another task that processes them. These tasks should run until a special message is received on either of the sources that causes everything to gracefully stop and the program to exit. Is there any similar sort of code that anyone has a link for that does this sort of thing?

Fabio Labella @SystemFw Apr 07 14:18
is the special message the same on both sources or different?
but basically use fs2-rabbit and fs2-kafka to get stream out of your sources (there are ways to write your own connectors ofc if you need them), then you can do:

rabbitStream
  .map(toCommonType)
  .merge(kafkaStream map toCommonType)
  .takeWhile(_ == specialMessage)
  .evalMap(yourProcessingTaskReturningIO)
  .compile
  .drain
  .as(ExitCode.Success)

Wrapping old Java code with FS2

One suggestion from Gitter was to use a tagless solution.

Oleg Pyzhcov @oleg-py Jun 27
The common pattern is to do something like this:

trait YoloContext[F[_]] {
  def execute(): F[Unit]
  def insert[T](o: T): F[Handle[T]]
}

object YoloContext {
  private class Implementation[F[_]: Sync](javaCtx: YoloCtx) extends YoloContext[F] { 
    /* implement methods. You have javaCtx here */ 
  }
  def create[F[_]: Sync](name: String): Resource[F, YoloContext[F]] = { 
    /* return a resource with `Implementation`. Hide the java stuff completely*/ 
  }
}

The algebra of streaming

This isn't FS2 specific but this salient comment from the omniscient Rob Norris is worth repeating:

Rob Norris @tpolecat Sep 23 22:58
Yeah in general it [fold] has to be lazy because you might try to fold something infinite.
And the thing that is not immediately intuitive is that foldLeft cannot possibly terminate for an infinite structure, but foldRight might.

Think about where the parens stack up. With foldLeft it looks like (((((((((( forever.

With foldRight it looks like a + (b + (c + ... forever.

So if + is lazy you might be able to stop after a while.