NOTE: This post requires previous knowledge of scala using the Akka framework specifically Akka streams.

Akka streams structure

The Akka streams module of Akka provides a beautiful DSL that will help you to move data in an asynchronous way. The base of an Akka stream is composed of 3 main components:

Source ~> Flow ~> Sink

• Source: Which is in charge of emitting the data through the stream, only have outputs ports
• Flow: receive the data, transform it and send it to the next component
• Sink: The end of the stream it will be the last step of any stream, only have input ports.

The problem

So far so good but what happens when we need to get the data from a source and this receives the key on the fly? i.e we need to download a file but we don't know the name of the file which arrives frow a previous step and we are using a library to download the file which is a source.

FlatMapConcat FTW

The FlatMapConcat flow allows us to transform any input element into a source this means we can connect a flow with a source, something like this:

Source ~> Flow ~> Source ~> Sink

A simple sample would look like this

1    //Simple Source
2    val src = Source(List(1, 2, 3))
3
4    val fromSrc = b.add(Flow[Int].log("from-Src"))
5
6    //Emit new list per element received
7    val flow = b.add(Flow[Int].flatMapConcat{ element =>
8      Source(List(4, 5, 6))
9    })
10
11    //¯\_(ツ)_/¯ nothing to do, just for sample purposes
12    val map = b.add(Flow[Int].map{ element =>
13      element
14    })
15
16    src ~> fromSrc ~>flow ~> map ~> sink

At line 16 in the main flow, we can see the 3rd step of our process is a regular flow but digging into it at line 8 we notice this flow is actually a Source that receives data from the flow FromSrc and emits new data to the map flow. 🙌🏻

That's it we just connected a source in the middle of a stream that emits new data based on the received data.

...
In case you want to run the sample check out my gist here