Offer Flow from Source => Source function

[derolf]

Assume we have a function of type

func[A, B](in: Source[A]) : Source[B]

These are for example being generated by the akka-grpc framework.

We would like to have a method to construct a Flow from func:

Flow.fromSource2Source[A, B](func: Source[A] => Source[B]) : Flow[A, B]

[2m]

You can use flatMapConcat with such function to achieve that:

val source = Source(0 to 5)
val duplicator: Int => Source[Int, NotUsed] = el => Source(List(el, el))
source.flatMapConcat(duplicator).runForeach(println)

https://scalafiddle.io/sf/v5ntXyA/0

[derolf]

I already found a solution:

def SourceProcessor[In, Out](f : Source[In, NotUsed] => Source[Out, NotUsed]): Flow[In, Out, NotUsed] =
  Flow[In].prefixAndTail(0).flatMapConcat { case (Nil, in) => f(in) }

It uses prefixAndTail to highjack the underyling Source.

[2m]

Ahh, I misunderstood. Good that you found a solution.

[derolf]

@2m Anyway, I appreciate your help! It took me half a day to figure out that “hack”...

[raboof]

Good find, that is definitely not trivial! Perhaps we should have a convenience method for this somewhere, or at least mention it as a pattern somewhere in the docs?

[derolf]

@raboof I basically started to write my own GraphStage until I found out that I am actually cloning prefixAndTail...

Yes, could be just added as FlowOps.via(func: (Source) => Source).