Search code examples
scalaakkaakka-stream

The meaning of the Keep combination?


I am trying to under the Keep combination in akka streams and created the following example:

import java.nio.file.Paths

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, IOResult}
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.util.ByteString

import scala.concurrent.Future
import scala.util.{Failure, Success}

object FileConsumer extends App {

  implicit val system = ActorSystem("reactive-tweets")
  implicit val materializer = ActorMaterializer()

  val source: Source[Int, NotUsed] = Source(1 to 100)
  val factorials = source.scan(BigInt(1))((acc, next) => acc * next)

  val result: Future[IOResult] =
    factorials.map(_.toString).runWith(lineSink("factorial2.txt"))

  implicit val ec = system.dispatcher
  result.onComplete {
    case Success(v) => println(s"Fileinfo ${ v.count }")
    case Failure(e) => println(e)
  }

  def lineSink(filename: String): Sink[String, Future[IOResult]] =
    Flow[String].map(s => ByteString(s + "\n")).toMat(FileIO.toPath(Paths.get(filename)))(Keep.right)


} 

on the akka streams website it says:

The resulting blueprint is a Sink[String, Future[IOResult]], which means that it accepts strings as its input and when materialized it will create auxiliary information of type Future[IOResult] (when chaining operations on a Source or Flow the type of the auxiliary information—called the “materialized value”—is given by the leftmost starting point; since we want to retain what the FileIO.toPath sink has to offer, we need to say Keep.right).

but what, when I want to keep the ByteString on the left side, I've tried:

  def lineSink2(filename: String): Sink[String, Future[ByteString]] =
Flow[String].map(s => ByteString(s + "\n")).toMat(Sink.foreach(println))(Keep.left)

but it does not compile at all.

I also do not understand:

is given by the leftmost starting point

Is leftmost starting point the Flow?

I think, I do not understand the idea of Keep yet.


Solution

  • Definition of Sink.foreach is as follows:

    def foreach[T](f: T ⇒ Unit): Sink[T, Future[Done]]
    

    That means materialised value is Future[Done]

    In case of flow you have:

     val value: Flow[String, ByteString, NotUsed] = Flow[String].map(s => ByteString(s + "\n"))
    

    its materialised value is NotUsed

    In this case:

    Keep.left - NotUsed - Materialised Value of Source or Flow

    Keep.right - Future[Done] - Materalised Value of Sink

    Keep.both - (NotUsed, Future[Done])

    Important fact is materialised value in many cases is NOT value of elements flowing through the stream, its rather

    • diagnostic information
    • stream state
    • other kind of information about stream