What you're trying to build here is pretty similar to "Balance",
so readint it's source may help you out:
https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/scaladsl/Graph.scala#L596
Also please note that the current impl:
- will eventually blow up with double-push or double-pull (one can not push
a port twice), in the current impl:
- the wiring seems wrong to me, it should specifically pick which output
port you're connecting from (`E.out(1) ~> ...`).
Can't solve the entire stage for you right now but hope this helped!
On Tue, Mar 15, 2016 at 2:40 PM, Ismail Kelebek <ismailkeleb...@gmail.com>
wrote:
> 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 akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Cheers,
Konrad 'ktoso' Malawski
Akka <http://akka.io/> @ Lightbend <http://lightbend.com/>
--
>>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
