I wanted to create a custom balance junction starting with the code below.
First of all it behaves different in debug and run mode. They both have sth
in common, they stop at 52. When i run it it gives this result :
run: debug :
0 nnn0
1 nnn1
2 nnn2
3 nnn3
... ...
... ...
... ...
52 nnn52
When i debug it , it starts with the out0's OutHandler, when i run it it
starts with the out1's OutHandler.
My main issue here is it stops after a pull event to a different
OutHandler. After the OutHandler block completed with the pull(in) method,
nothing arrives at the InHandler of inlet in.
class MyBalance extends GraphStage[UniformFanOutShape[Int, Int]] {
val in = Inlet[Int]("myBalance.in")
val out0 = Outlet[Int]("myBalance.out0")
val out1 = Outlet[Int]("myBalance.out1")
val out2 = Outlet[Int]("myBalance.out2")
val out3 = Outlet[Int]("myBalance.out3")
val shape = UniformFanOutShape(in,out0,out1,out2,out3)
override def createLogic(inheritedAttributes: Attributes):
GraphStageLogic =
new GraphStageLogic(shape) {
var whichOutlet = 0
setHandler(in, new InHandler {
override def onPush(): Unit = {
val elem = grab(in)
whichOutlet match {
case 0 => push(out0, elem)
case 1 => push(out1, elem)
case 2 => push(out2, elem)
case 3 => push(out3, elem)
case _ => println("wrong outlet")
}
}
})
setHandler(out0, new OutHandler {
override def onPull(): Unit = {
whichOutlet = 0
pull(in)
}
})
setHandler(out1, new OutHandler {
override def onPull(): Unit = {
whichOutlet = 1
pull(in)
}
})
setHandler(out2, new OutHandler {
override def onPull(): Unit = {
whichOutlet = 2
pull(in)
}
})
setHandler(out3, new OutHandler {
override def onPull(): Unit = {
whichOutlet = 3
pull(in)
}
})
}
}
val runnableGraphBalanceTest =
RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>
val A: Outlet[Int] = builder.add(Source((0 to
1000).toList)).out
val E: UniformFanOutShape[Int, Int] = builder.add(balanceGraphStage)
A ~> E ~> Sink.foreach[Int](x => println("nnn" + x))
E ~> Sink.foreach[Int](x => println(x))
E ~> Sink.foreach[Int](x => println("hhhhhhhhh" + x))
E ~> Sink.foreach[Int](x => println("rrrrrrrrrrrrrrrrrrrr" +
x))
ClosedShape
})
runnableGraphBalanceTest.run
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
