Akka Stream

[jimmysaints93]

Feb 10, 2020

Topic: Streaming all the things with Akka Streams by Johan Andrén Link: https://www.youtube.com/watch?v=jTUkmvPMlUE

Reactive Streams

• async stream processing
• non-blocking back pressure

Example Scenario:

Source -> 10 msg/s (fast producer) --async boundary-- Flow Sink <- 1 msg/s (slow consumer)

results in OutOfMemoryError, no back pressure

Q: How can we guarantee that in this scenario, source is notified that the data as been consumed? A: Async non blocking back pressure, downstream (Sink to flow) will always signal upstream Example: Sink's buffer can cater 2 elements it'll force the fast producer to slow down.

implicit val system = ActorSystem() implicit val mat = ActorMaterializer()

val source = Source(0 to 20000000) Int val flow = Flow[Int].map(.toString()) Int -> String val sink = Sink.foreach[String](println()) String val runnable = source.via(flow).to(sink) runnable.run()

or

Source(0 to 200000) .map(_.toString) .runForeach(println)

• in TCP both end has buffer
• WebSocket - 2 way flow (upgraded HTTP)
• in webSocket, there are 2 types of message, Strict (1 chunk message) vs Streamed Message(Many chunks)

Pancake methaphor:

battersource --> pan 1 -> pan 2 -> hungrysink

-async has a cost

val fryingPan1: Flow[ScoopOfBatter, HalfCookedPancake, NotUsed] = Flow[HalfCookedPancake].map { halfCooked = > Pancake() } val fryingPan2: Flow[HalfCookedPancake, Pancake, NotUsed] = Flow[HalfCookedPancake].map { halfCooked = > Pancake() }

val pancakeChef: Flow[ScoopOfBatter, Pancake, NotUsed] = Flow[ScoopOfBatter, Pancake, NotUsed] = Flow[ScoopOfBatter].via(fryingPan1.async).via(fryingPan2.async)

FORK JOIN? (easy to scale, add more frying pans, increase the throughput, bad when requires order, complete task a first then task b )

val pancakeChef: FLow[ScoopOfBatter, Pancake, NotUsed] = Flow.fromGraph(GraphDSL.create()) { implicit builder => import GraphDSL.IMplicits._

val dispatchBatter = builder.add(Balance[ScoopofBatter](2))
val mergePancakes = builder.add(Merge[Pancake](2))

dispatchBatter.out(0) ~-> fryingPan1.async ~ fryingPan2.async ~> mergePancakes.in(0)
dispatchBatter.out(1) ~-> fryingPan1.async ~ fryingPan2.async ~> mergePancakes.in(1)

FLowSHape(dispatchBatter.in, mergePancakes.out)

}

pipeline vs parallel

Alpakka - modern camel???

GraphStage API - for unique snowflake use cases

todo:

• research (KAFKA)
• research dispatcher

default dispatcher - fork join pool

materializer

• graph blueprint, blueprint to execute, house -> money
• blueprint to running stream
• async island (map map map) create sm island run on top of actor and fuse
• has lifecycle
• bunch of streams that have their lifecycle to one materializer
• more expensive that regular objects
• thousands are expensive
• runs on top of actor system
• forgetting to close will leak

https://github.com/johanandren/akka-stream-samples/tree/jfokus-2017 pub sub reactive streams api java rx, reactor, vertix?

[jimmysaints93]

stream-graphs: https://doc.akka.io/docs/akka/2.5/stream/stream-graphs.html

[jimmysaints93]

• When constructing flows and graphs in Akka Streams think of them as preparing a blueprint, an execution plan
• Stream materialization is the process of taking a stream description (RunnableGraph) and allocating all the necessary resources it needs in order to run.
• Materialization is triggered at so called “terminal operations”.
• Materialization is currently performed synchronously on the materializing thread.

[jimmysaints93]

Async boundary example:

• Source(List(1, 2, 3)).map( + 1).async.map( * 2).to(Sink.ignore)

Diagram:

• https://doc.akka.io/docs/akka/2.5/images/asyncBoundary.png

[jimmysaints93]

Reactive Streams semantics

emits when the mapping function returns an element or there are still remaining elements from the previously calculated collection

backpressures when downstream backpressures or there are still available elements from the previously calculated collection

completes when upstream completes and all remaining elements has been emitted

operators are Source, Flow, Sink, Fan-In, Fan-Out, BidiFlow

mapConcat: https://doc.akka.io/docs/akka/current/stream/operators/Source-or-Flow/mapConcat.html transform each element into zero or more elements that are individually passed downstream,

[jimmysaints93]

Stream composition:

https://doc.akka.io/docs/akka/current/stream/stream-composition.html

• BidiFlow operators are usually useful in IO related tasks, where there are input and output channels to be handled
• The TLS support in Akka is for example implemented as a BidiFlow

[jimmysaints93]

futures & promises: http://dist-prog-book.com/chapter/2/futures.html

[jimmysaints93]

Challenge: 1 input, N outputs, Solution: combining 1 to N (broadcast)

https://stackoverflow.com/questions/44926580/how-to-combine-2-sinks-of-a-different-type

Fan-out approach: https://doc.akka.io/docs/akka/current/stream/stream-graphs.html

[jimmysaints93]

// QUICK AKKA STREAM SAMPLE

val sourcePath = Paths.get("src/main/resources/db/source.txt")
val sampleSource = FileIO.fromPath(sourcePath)

val sinkPath = Paths.get("src/main/resources/db/sink.txt")
val sampleSink = FileIO.toPath(sinkPath)

val sampleRunnableGraph: RunnableGraph[Future[IOResult]] = sampleSource.to(sampleSink)
sampleRunnableGraph

sampleRunnableGraph.run().foreach( result =>
  println(s"result status: ${result.status}, ${result.count} bytes read")
)

// END OF QUICK AKKA STREAM SAMPLE

[jimmysaints93]

Good websocket examples: https://github.com/akka/akka-http/blob/v10.1.11/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala

[jimmysaints93]

WebSocket explanation: https://www.ably.io/concepts/websockets

[jimmysaints93]

WebSocket specs: https://tools.ietf.org/html/rfc6455

[jimmysaints93]

explantion of via/viaMat/to/toMat operators:

https://stackoverflow.com/questions/37911174/via-viamat-to-tomat-in-akka-stream

source.to(sink)    equivalent to   source.toMat(sink)(Keep.left)
flow1.via(flow2)   equivalent to   flow1.viaMat(flow2)(Keep.left)

[jimmysaints93]

https://stackoverflow.com/questions/53437098/triggering-on-fileio-topath-sink-completion

somehow, combining 2 sinks:

1. connect source to sink1
2. materialize the value of sink1 using "alsoToMat" & "Keep.right"
3. map the materialized value, in case of FileIO.toPath, a Future[IOResult]

[jimmysaints93]

Good documentation for Akka Streams: https://svezfaz.github.io/akka/streams/reactive/2017/03/27/on-akka-streams-stages-1.html

[jimmysaints93]

Good slide show for quick Akka Stream explanation:

HttpServer as a:
Flow[HttpRequest, HttpResponse]

HttpEntity as a:
Source[ByteString, _]

Websocket connection as a:
Flow[ws.Message, ws.Message]