Akka stream with a tree structure

Question

I have a flow:

A +----> B +----> D
  |        |
  |        +----> E
  |
  +----> C +----> G
           |
           +----> H

I want implement this flow by Akka Stream. Output of Node B and C is a List and input of D,E,G,H is an element in the List. I have tried: - Create a Source from Sink of Flow A ---> B or A --->C

But i am not lucky. Do you have any suggestion?

Possible duplicate of [How to convert a tree structure in a stream of nodes in java](https://stackoverflow.com/questions/26158082/how-to-convert-a-tree-structure-in-a-stream-of-nodes-in-java) — Raman Mishra, Jan 11 '19 at 06:06
@RamanMishra: Thanks for your suggestion but I want to implement this tructure on Akka Stream. — Toan Nguyen, Jan 11 '19 at 06:12
you can have a source two flow and for each flow 2 different sink what is the problem? — Raman Mishra, Jan 11 '19 at 06:17
My problem is i cann't create a Source from Sink or Flow of A--->B and A--->C — Toan Nguyen, Jan 11 '19 at 06:20
Where you need to create Source from sink? B and C will be flow and D, E and G, H will be sink @Toan Nguyen — Raman Mishra, Jan 11 '19 at 06:37
I need to create Source from Sink of B and C flow. and This Source is for D,E and G,H flow. — Toan Nguyen, Jan 11 '19 at 06:46
Have you tried any code? to put in your question a see what happens.. — Emiliano Martinez, Jan 11 '19 at 09:25
What you are trying to do should be achievable using graphs. You can learn more here: https://doc.akka.io/docs/akka/2.5/stream/stream-graphs.html — emilianogc, Jan 11 '19 at 11:52

score 0 · Answer 1 · answered Jan 14 '19 at 16:08

As mentioned in the comments, you need to use the graph functions: https://doc.akka.io/docs/akka/2.5/stream/stream-graphs.html

For instance after A, you need to add a Broadcaster with 2 outputs to pass the messages to both B and C. Same after B and C.

val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
  import GraphDSL.Implicits._
  val a = Source(1 to 10)

  val bcast = builder.add(Broadcast[XXX](2))
  val b = builder.add(...)
  val c = builder.add(...)

  a ~> bcast ~> b
  a ~> bcast ~> c
  ...
})

Akka stream with a tree structure

1 Answers1