The purpose of the series is to uncover some of theAkka Streams essentials. There are few major principles that lie behinds this topic. They are:
- Asynchronous
- Backpressured
- Incremental
- And potentially infinite STREAMS of elements
These terms are inextricably linked to the Reactive Streams concepts:
Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure. This encompasses efforts aimed at runtime environments (JVM and JavaScript) as well as network protocols.
reactive-streams.org
In other words, Reactive Streams specifies the behavior of asynchronous backpressured streams.
Reactive Streams concepts
There are some basic concepts that shape Reactive Streams:
- Publisher: asynchronously emits elements
- Subscriber: receives elements
- Processor: transforms elements
- Async: every piece of data flows asynchronously
- Backpressure: publishers do not overwhelm subscribers
By the way, Reactive Streams is rather a service provider interface (SPI) than an API. Libraries that implement the specification are able to choose their own API and Akka Streams is just one of them.
The main building blocks of asynchronous streams in Akka are:
These components are usually backed by Akka actors. We build a reactive stream by connecting the components together as depicted on the next image:
There are two directions in graphs that can be defined. The first one called “downstream” is for the elements which flow towards the sink. The second one, in opposition to the first, is called the “upstream” and guides towards the source.
So let’s finally get our hands dirty and look at the real code examples.
Hands-on
Now let’s create a new project to discover some Akka Streams essentials under the hood. I use Intellij IDEA and sbt as a build tool. We have to define some Akka-related dependencies in the build.sbt:
| name := "akka-streams" | |
| version := "0.1" | |
| scalaVersion := "2.12.10" | |
| lazy val akkaVersion = "2.5.19" | |
| lazy val scalaTestVersion = "3.0.5" | |
| libraryDependencies ++= Seq( | |
| "com.typesafe.akka" %% "akka-stream" % akkaVersion, | |
| "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion, | |
| "com.typesafe.akka" %% "akka-testkit" % akkaVersion, | |
| "org.scalatest" %% "scalatest" % scalaTestVersion | |
| ) |
Then let’s create a new Scala object called FirstPrinciples under any package. It will act as a sandbox for further testing. Don’t forget to extend the App trait to make the class runnable.
Since Akka Streams, as it comes from its name, uses Akka actors
behind the scenes, we have to create an Actor System beforehand. The second important step is to create an ActorMaterializer which is needed to run Akka Streams graph (to “materialize” them):
| implicit val system = ActorSystem("FirstPrinciples") | |
| implicit val materializer = ActorMaterializer() |
We will take a closer look at materializer in the next article from the series.
As we’ve mentioned before, there are three basic concepts in Akka Streams: the source, the sink, and the flow. Basically, any graph may not contain any flows, emitting elements directly from a source to a sink:
| // source | |
| val source = Source(1 to 10) | |
| // sink | |
| val sink = Sink.foreach[Int](println) | |
| // a graph | |
| val graph: RunnableGraph[NotUsed] = source.to(sink) | |
| graph.run() // prints 1 to 10 |
Here the source emits a limited collection of Int from 1 to 10. The sink uses its foreach method to execute the println function for each element emitted from the source. Pay attention that nothing will happen until the run() method is called. The example Akka Streams graph prints elements from 1 to 10 as they emitted from the source.
Flow concept in Akka Streams used to make arbitrary transformations on the incoming elements stream. A good example is map which applies some function on each element from a stream:
| // flows – transform elements | |
| val flow = Flow[Int].map(x => x + 1) | |
| val sourceWithFlow = source.via(flow) | |
| val flowWithSink = flow.to(sink) | |
| // combine them | |
| sourceWithFlow.to(sink).run() // prints 2 to 11 | |
| source.to(flowWithSink).run() // prints 2 to 11 | |
| source.via(flow).to(sink).run() // prints 2 to 11 |
There is the map function at line #2 which multiplies each Int from the stream by 2.
There is one more thing to note: Null is not allowed as a stream element, use Scala Option instead. In the following example the code won’t compile due to exception:
| // NULLs are not allowed | |
| val illegalSource = Source.single[String](null) |
Exception in thread “main” java.lang.NullPointerException: Element must not be null, rule 2.13
Source types
There are few types of sources in Akka Streams:
| // kinds of sources | |
| val finiteSource = Source.single(1) | |
| val anotherfiniteSource = Source(List(1, 2, 3)) | |
| val emptySource = Source.empty[Int] | |
| val infiniteSource = Source(Stream.from(1)) | |
| import scala.concurrent.ExecutionContext.Implicits.global | |
| val futureSource = Source.fromFuture(Future(42)) |
They are mostly self-explanatory and don’t need any detailed clarification.
Sink types
There are also a few sink types:
| // sinks | |
| val doNothingSink = Sink.ignore | |
| val foreachSink = Sink.foreach[String](println) | |
| val headSink = Sink.head[Int] // retrieves head and closes the stream | |
| val foldSink = Sink.fold[Int, Int](0)((a, b) => a + b) |
The most interesting here is the foldSink. It folds over the emitted element with a function, where each invocation will get the new element and the result from the previous fold invocation. The first invocation will be provided the zero value. So let’s take a look at the following code snippet:
| // fold sink | |
| val foldSource = Source[Int](1 to 10) | |
| val foldTestSink = Sink.fold[Int, Int](0)(_ + _) | |
| foldSource.to(foldTestSink).run() |
The result will show the sum of all elements in the stream.
Flow types
There are also some subtypes of Flow are available:
| // flows | |
| val mapFlow = Flow[Int].map(x => 2 * x) | |
| val takeFlow = Flow[Int].take(5) // only take the first five elements | |
| // drop, filter | |
| // doesn't have flatMap |
Let’s combine them in a graph:
| // source – flow – flow – sink | |
| val doubleFlowGraph = source.via(mapFlow).via(takeFlow).to(sink) | |
| doubleFlowGraph.run() |
And run it:
2
4
6
8
10
The graph takes an input source, multiply each element by two, and then takes the first five elements from the result.
Test task
Now let’s consider the following task:
Create a stream that takes the names of persons and then keep the first two names with length > 5 characters.
The solutions are shown at the code snippet below:
| val names = List("Alex", "John", "Daniel", "Jeremy", "Jonathan", "Matt") | |
| val namesSource = Source[String](names) | |
| val filterFlow = Flow[String].filter(name => name.length > 5) | |
| val takeTwoFlow = Flow[String].take(2) |
Compose the graph and run:
| // compose the graph | |
| namesSource.via(filterFlow).via(takeTwoFlow).runForeach(println) |
The result is:
Daniel
Jeremy
Conclusion
In the article, we’ve observed the very basics of building graphs in Akka Streams. We’ve learned how the elements flow between graph stages, which essential types each concept has, and how to compose all-together in a runnable graph.