Endre- Apologies, but I just saw your response as a result of wwagner4's inquiry.
Thanks for reading the post and I'm glad you like the animation - it was done in Animatron. That's manual animation, brother :) No luck finding an easier solution. I'm putting together a talk on Akka Streams / HTTP and wanted a good way to visualize streams. Good point - I'll be sure to note the immediate demand signal in my talk, though I think the simplified synchronous model lends conceptual clarity to those approaching streams for the first time. Thanks for the feedback! -Lance On Tuesday, June 2, 2015 at 4:21:00 AM UTC-4, Akka Team wrote: > > Btw, small nitpick: The animation shows what I call the "synchronous" > execution model, where the demand starts from the end of the stream and > travels back and forth (branching out if necessary). In reality, these > stages individually demand the first element when they are started, and > they immediately request the next element once they started working on a > received element. > > -Endre > > On Tue, Jun 2, 2015 at 10:10 AM, Akka Team <[email protected] > <javascript:>> wrote: > >> Wow! >> >> What did you use for the animation? I want that :D >> >> -Endre >> >> On Tue, Jun 2, 2015 at 12:10 AM, Lance Arlaus <[email protected] >> <javascript:>> wrote: >> >>> Circling back on this, I created a blog post that explains the issue I >>> encountered along with the solution of using a balancing buffer. >>> >>> >>> http://blog.lancearlaus.com/akka/streams/scala/2015/05/27/Akka-Streams-Balancing-Buffer/ >>> >>> I hope it helps those who encounter the same issue. >>> >>> >>> On Saturday, May 9, 2015 at 11:49:48 AM UTC-4, Lance Arlaus wrote: >>>> >>>> No problem. >>>> Thanks for the quick response and here's the corresponding issue: >>>> https://github.com/akka/akka/issues/17435 >>>> >>>> On Saturday, May 9, 2015 at 5:05:44 AM UTC-4, drewhk wrote: >>>>> >>>>> Hi Lance, >>>>> >>>>> On Sat, May 9, 2015 at 12:49 AM, Lance Arlaus <[email protected]> >>>>> wrote: >>>>> >>>>>> Hi- >>>>>> >>>>>> I've encountered an issue with processing a stream that I fan out via >>>>>> broadcast and fan in via zip. >>>>>> The broadcast splits the stream in two with one branch containing a >>>>>> drop element. >>>>>> According to my read of the docs, I would expect the terminating zip >>>>>> to complete when the shorter of the two streams (the one with the drop) >>>>>> completes. >>>>>> However, the flow hangs waiting indefinitely. >>>>>> >>>>>> Here's the relevant part of a test case I put together to reproduce >>>>>> the problem. >>>>>> Note that the flow without the drop (the first flow) works fine with >>>>>> different length streams. >>>>>> What am I doing wrong? >>>>>> >>>>> >>>>> I don't think you are doing anything wrong. Btw, I suspect the bug >>>>> being in Broadcast instead. Can you file a ticket please? >>>>> >>>>> -Endre >>>>> >>>>> >>>>>> >>>>>> Akka Stream Version: 1.0-RC2 >>>>>> >>>>>> Thanks, >>>>>> Lance >>>>>> >>>>>> // This flow works fine >>>>>> def zipSource(num: Int, diff: Int) = Source() { implicit b => >>>>>> import akka.stream.scaladsl.FlowGraph.Implicits._ >>>>>> >>>>>> val source0 = b.add(Source(1 to num)) >>>>>> val source1 = b.add(Source(1 to (num + diff))) >>>>>> val zip = b.add(Zip[Int, Int]) >>>>>> >>>>>> source0 ~> zip.in0 >>>>>> source1 ~> zip.in1 >>>>>> >>>>>> (zip.out) >>>>>> } >>>>>> >>>>>> // This flow waits indefinitely when diff > 0 >>>>>> def zipDropSource(num: Int, diff: Int) = Source() { implicit b => >>>>>> import akka.stream.scaladsl.FlowGraph.Implicits._ >>>>>> >>>>>> val source = b.add(Source(1 to (num + diff))) >>>>>> val bcast = b.add(Broadcast[Int](2)) >>>>>> val drop = b.add(Flow[Int].drop(diff)) >>>>>> val zip = b.add(Zip[Int, Int]) >>>>>> >>>>>> source ~> bcast ~> zip.in0 >>>>>> bcast ~> drop ~> zip.in1 >>>>>> >>>>>> (zip.out) >>>>>> } >>>>>> >>>>>> // PASS >>>>>> "Zip" should "complete with same length streams" in { >>>>>> val future: Future[Int] = zipSource(10, >>>>>> 10).runWith(Sink.fold(0)((s, i) => s + 1)) >>>>>> whenReady(future)(_ shouldBe 10) >>>>>> } >>>>>> >>>>>> // PASS >>>>>> it should "complete with different length streams" in { >>>>>> val future: Future[Int] = zipSource(10, >>>>>> 20).runWith(Sink.fold(0)((s, i) => s + 1)) >>>>>> whenReady(future)(_ shouldBe 10) >>>>>> } >>>>>> >>>>>> // PASS >>>>>> "Zip with drop" should "complete with same length streams" in { >>>>>> val future: Future[Int] = zipDropSource(10, >>>>>> 0).runWith(Sink.fold(0)((s, i) => s + 1)) >>>>>> whenReady(future)(_ shouldBe 10) >>>>>> } >>>>>> >>>>>> // FAIL >>>>>> it should "complete with different length streams" in { >>>>>> val future: Future[Int] = zipDropSource(10, >>>>>> 10).runWith(Sink.fold(0)((s, i) => s + 1)) >>>>>> whenReady(future)(_ shouldBe 10) >>>>>> } >>>>>> >>>>>> } >>>>>> >>>>>> -- >>>>>> >>>>>>>>>> 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 http://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] <javascript:>. >>> To post to this group, send email to [email protected] >>> <javascript:>. >>> Visit this group at http://groups.google.com/group/akka-user. >>> For more options, visit https://groups.google.com/d/optout. >>> >> >> >> >> -- >> Akka Team >> Typesafe - Reactive apps on the JVM >> Blog: letitcrash.com >> Twitter: @akkateam >> > > > > -- > Akka Team > Typesafe - Reactive apps on the JVM > Blog: letitcrash.com > Twitter: @akkateam > -- >>>>>>>>>> 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 http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
