Thanks a lot for your informations (and sorry for the double post). Can you please tell me what is the way to "wake up the stage", it is not really obvious how to make that happen.
Thanks Le vendredi 14 octobre 2016 04:09:12 UTC-4, Konrad Malawski a écrit : > > Please read this: > - http://blog.akka.io/integrations/2016/08/29/connecting-existing-apis > - and this: > http://doc.akka.io/docs/akka/2.4/scala/stream/stream-customize.html > > Specifically, your trigger should be implemented as async-callback, as it > comes from the outside but should "wake up" the stage to be able to push > data again. > In your current setup it never "wakes up" since all pulls/pushes have been > processed - the stage has no idea it should do something once you called > open. > > On Fri, Oct 14, 2016 at 12:13 AM, regis leray <regis...@gmail.com > <javascript:>> wrote: > >> Hi, >> >> Im currently trying to implement a valve Graph to manage pause/resume. We >> can control the behavior of the graph by using the MaterializeValue >> >> trait ValveSwitch { >> def open: Unit >> def close: Unit >> } >> >> >> Current implementation of the valve >> >> class Valve[A](mode: ValveMode = ValveMode.Open) extends >> GraphStageWithMaterializedValue[FlowShape[A, A], ValveSwitch] { >> >> override val shape = FlowShape(Inlet[A]("valve.in"), >> Outlet[A]("valve.out")) >> >> override def createLogicAndMaterializedValue(inheritedAttributes: >> Attributes): (GraphStageLogic, ValveSwitch) = { >> val logic = new ValveGraphStageLogic(shape, mode) >> (logic, logic.switch) >> } >> >> } >> >> >> The current implementation is pretty simple, each time we are receiving a >> onPull demand we are requesting by doing pull(in). >> When a onPush demand is received we are checking the current state >> - if Open we are doing the default behavior by doing push(out,element) >> - if Close we are putting the element into a queue >> >> private class ValveGraphStageLogic(shape: Shape, var mode: ValveMode) >> extends GraphStageLogic(shape){ >> import shape._ >> >> var bufferedElement = List.empty[A] >> >> val switch = new ValveSwitch { >> override def open: Unit = { >> mode = ValveMode.Open >> println(s"pushing $bufferedElement, out is available ? >> ${isAvailable(out)}") >> >> bufferedElement.foreach(push(out, _)) >> bufferedElement = List.empty >> } >> >> override def close: Unit = { >> mode = ValveMode.Closed >> } >> } >> >> setHandler(in, new InHandler { >> override def onPush(): Unit = { >> val element = grab(in) //acquires the element that has been received >> during an onPush >> println(s"${mode} on push called with $element") >> if (mode == ValveMode.Open) { >> push(out, element) //push directly the element on the out port >> } else { >> bufferedElement = bufferedElement :+ element >> } >> } >> }) >> >> setHandler(out, new OutHandler { >> override def onPull(): Unit = { >> println("on pull called") >> pull(in) //request the next element on in port >> } >> }) >> } >> >> >> When we are resuming the valve my using the switch.open, we are pushing >> the element >> >> override def open: Unit = { >> >> mode = ValveMode.Open >> println(s"pushing $bufferedElement, out is available ? >> ${isAvailable(out)}") >> >> bufferedElement.foreach(push(out, _)) >> bufferedElement = List.empty >> } >> >> >> The Current test is failing >> >> "A closed valve" should "emit only 3 elements after it has been open" in { >> >> val (valve, probe) = Source(1 to 5) >> .viaMat(new Valve(ValveMode.Closed))(Keep.right) //the current valve by >> default is closed, dont push any message >> .toMat(TestSink.probe[Int])(Keep.both) >> .run() >> >> probe.request(2) >> probe.expectNoMsg() >> >> valve.open //open the valve should emit the previous >> >> >> probe.expectNext shouldEqual 1 //we never receive the element >> probe.expectNext shouldEqual 2 >> >> probe.request(3) >> probe.expectNext shouldEqual 3 >> probe.expectNext shouldEqual 4 >> probe.expectNext shouldEqual 5 >> >> probe.expectComplete() >> } >> >> >> Here the console log >> >> on pull called >> Closed on push called with 1 >> pushing Some(1), out is available ? true >> >> Expected OnNext(_), yet no element signaled during 3 seconds >> java.lang.AssertionError: Expected OnNext(_), yet no element signaled during >> 3 seconds >> at >> akka.stream.testkit.TestSubscriber$ManualProbe.expectNext(StreamTestKit.scala:268) >> at >> akka.stream.testkit.TestSubscriber$ManualProbe.expectNext(StreamTestKit.scala:259) >> at >> com.omsignal.omrun.orchestration.rest.ValveSpec$$anonfun$1.apply(ValveSpec.scala:44) >> at >> com.omsignal.omrun.orchestration.rest.ValveSpec$$anonfun$1.apply(ValveSpec.scala:34) >> >> >> I'm suspecting the current code to have an issue when we are resuming the >> valve, it doesnt seems the push really works >> >> val switch = new ValveSwitch { >> override def open: Unit = { >> mode = ValveMode.Open >> println(s"pushing $bufferedElement, out is available ? >> ${isAvailable(out)}") >> >> bufferedElement.foreach(push(out, _)) >> bufferedElement = Option.empty >> } >> >> override def close: Unit = { >> mode = ValveMode.Closed >> } >> } >> >> >> There is definitively something i dont catch up, if anyone could help me >> to see some light.... >> >> Here the gist >> https://gist.github.com/regis-leray/013dfe030159bcd890ca0d5cd440c938 >> >> Any help would be appreciated >> >> -- >> >>>>>>>>>> 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+...@googlegroups.com <javascript:>. >> To post to this group, send email to akka...@googlegroups.com >> <javascript:>. >> 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/> @ Typesafe <http://typesafe.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.
