Oh, Actually it's because the source (Source(1 to 10)) is consumed automatically before consumer joins. I changed it to Source.tick(1.second, 1.second, 1) and everything works as expected.
On Fri, Oct 28, 2016 at 11:48 PM, auxdx <[email protected]> wrote: > Hi, > > I don't get it. I do not need to import source right? (it throws the error > if I do so: b.add(source)). In addition, if I am using s1 ~> > Sink.foreach(println) instead of source ~> Sink.foreach(println) then the > code works. > > Could you please elaborate more? > > I want to use BroadCastHub as a GraphStage instead so that I can do > something like: s1 ~> BroadCastHub ~> sink1; BroadCastHub ~> sink2; Is this > possible? The problem is that BroadCastHub when is materialised becomes a > Source; not GraphStage whereas what I want is GraphStage such as > BroadCast/Balance (akka built-in stages), etc... The reason is that I have > already had a CloseShape graph; and want to add BroadCastHub into the > existing CloseShape Graph. > > Thanks. > > On Friday, 28 October 2016 18:45:03 UTC+8, drewhk wrote: >> >> Hi, >> >> You have not added your Source to the Graph, so it will not work. You >> need b.add() to import things into the graph to wire them up. >> >> -endre >> >> On Fri, Oct 28, 2016 at 12:39 PM, auxdx <[email protected]> wrote: >> >>> Hi, >>> >>> I am wondering why this code does not work as intended: >>> >>> def testGraphBroadCastShape() = { >>> implicit val as = ActorSystem() >>> implicit val ec = as.dispatcher >>> val settings = ActorMaterializerSettings(as) >>> implicit val mat = ActorMaterializer(settings) >>> >>> val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit b => >>> import GraphDSL.Implicits._ >>> >>> //Importing the partial shape will return its shape (inlets & outlets) >>> val s1: Source[Int, NotUsed] = Source(1 to 10) >>> val source = s1.toMat(BroadcastHub.sink(bufferSize = >>> 256))(Keep.right).run() >>> >>> source ~> Sink.foreach(println) >>> >>> ClosedShape >>> }) >>> >>> g.run() >>> } >>> >>> >>> Why I cann't connect source to Sink.foreach? >>> >>> >>> Thanks. >>> >>> >>> >>> >>> -- >>> >>>>>>>>>> Read the docs: http://akka.io/docs/ >>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>> urrent/additional/faq.html >>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>> p/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. >>> >> >> -- > >>>>>>>>>> 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 a topic in the > Google Groups "Akka User List" group. > To unsubscribe from this topic, visit https://groups.google.com/d/ > topic/akka-user/QerpAGiuw-Q/unsubscribe. > To unsubscribe from this group and all its topics, 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. > -- >>>>>>>>>> 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.
