Defining and running streams基础

7.5.2 Defining and running streams

Linear processing pipelines can be expressed in Akka Streams using the following core abstractions:

Source A processing stage with exactly one output, emitting data elements whenever downstream processing
stages are ready to receive them.
Sink A processing stage with exactly one input, requesting and accepting data elements possibly slowing down
the upstream producer of elements
Flow A processing stage which has exactly one input and output, which connects its up- and downstreams by
transforming the data elements flowing through it.
RunnableGraph A Flow that has both ends “attached” to a Source and Sink respectively, and is ready to be
run().

It is possible to attach a Flow to a Source resulting in a composite source, and it is also possible to prepend a
Flow to a Sink to get a new sink. After a stream is properly terminated by having both a source and a sink, it
will be represented by the RunnableGraph type, indicating that it is ready to be executed.

It is important to remember that even after constructing the RunnableGraph by connecting all the source,
sink and different processing stages, no data will flow through it until it is materialized. Materialization is the
process of allocating all resources needed to run the computation described by a Graph (in Akka Streams this
will often involve starting up Actors). Thanks to Flows being simply a description of the processing pipeline they
are immutable, thread-safe, and freely shareable, which means that it is for example safe to share and send them
between actors, to have one actor prepare the work, and then have it be materialized at some completely different
place in the code.

val source = Source(1 to 10)
val sink = Sink.fold[Int, Int](0)(_ + _)
// connect the Source to the Sink, obtaining a RunnableGraph
val runnable: RunnableGraph[Future[Int]] = source.toMat(sink)(Keep.right)
// materialize the flow and get the value of the FoldSink
val sum: Future[Int] = runnable.run()

After running (materializing) the RunnableGraph[T] we get back the materialized value of type T. Every
stream processing stage can produce a materialized value, and it is the responsibility of the user to combine them
to a new type. In the above example we used toMat to indicate that we want to transform the materialized value
of the source and sink, and we used the convenience function Keep.right to say that we are only interested in
the materialized value of the sink. In our example the FoldSink materializes a value of type Future which will
represent the result of the folding process over the stream. In general, a stream can expose multiple materialized
values, but it is quite common to be interested in only the value of the Source or the Sink in the stream. For
this reason there is a convenience method called runWith() available for Sink, Source or Flow requiring,
respectively, a supplied Source (in order to run a Sink), a Sink (in order to run a Source) or both a Source
and a Sink (in order to run a Flow, since it has neither attached yet).

val source = Source(1 to 10)
val sink = Sink.fold[Int, Int](0)(_ + _)
// materialize the flow, getting the Sinks materialized value
val sum: Future[Int] = source.runWith(sink)

It is worth pointing out that since processing stages are immutable, connecting them returns a new processing
stage, instead of modifying the existing instance, so while constructing long flows, remember to assign the new
value to a variable or run it:Ï

val source = Source(1 to 10)
source.map(_ => 0) // has no effect on source, since it’s immutable
source.runWith(Sink.fold(0)(_ + _)) // 55
val zeroes = source.map(_ => 0) // returns new Source[Int], with ‘map()‘ appended
zeroes.runWith(Sink.fold(0)(_ + _)) // 0