Hi Wolfgang, What you see is completely expected. The strategy you added only defines the strategy for that particular consumer. Once you attach other processing steps, like those "map"s they do their own batching strategy independent from what your consumer does. This is by design, backpressure scheduling strategies are not transitive (and should never be actually).
You can imagine your stages like freely combinable elements: [ActorPublisher] [--> (map buffer+schedule) --> map] [--> (buffer+WatermarkStrategy) --> yourActor] When you assemble them like this: [ActorPublisher] [--> (buffer+WatermarkStrategy) --> yourActor] then of course your publisher will observe the schedule of your choosing. But if you assemble: [ActorPublisher] [--> (map buffer+schedule) --> map] [--> (buffer+WatermarkStrategy) --> yourActor] Then obviously your publisher will observe the schedule of the map directly, while the map will observe the schedule of your consumer. The backpressure protocol is always local between a producer-consumer pair. -Endre On Mon, Dec 22, 2014 at 11:59 AM, Wolfgang Friedl < wolfgang.fri...@hotmail.com> wrote: > > Following Sample: > > The "Producer" receives a different demand of elements dependent if the > flow does have processing steps or not. > (*Consumer uses the WatermarkStrategy(30)*) > > > var flow = Flow.empty[Int] > > > flow.runWith(PublisherSource(ActorPublisher[Int](intProducer)),SubscriberSink(ActorSubscriber[Int](intConsumer))) > > -->Requested Demand in the Producer > > *30* var flow = Flow.empty[Int] > flow = flow.map(_ +1).map(_ *2) > > > flow.runWith(PublisherSource(ActorPublisher[Int](intProducer)),SubscriberSink(ActorSubscriber[Int](intConsumer))) > > -->Requested Demand in the Producer *4* > > > > > (*Consumer uses the OneByOneStrategy*) > > > var flow = Flow.empty[Int] > flow = flow.map(_ +1).map(_ *2) > > > flow.runWith(PublisherSource(ActorPublisher[Int](intProducer)),SubscriberSink(ActorSubscriber[Int](intConsumer))) > > -->Requested Demand in the Producer > > *4* var flow = Flow.empty[Int] > > > flow.runWith(PublisherSource(ActorPublisher[Int](intProducer)),SubscriberSink(ActorSubscriber[Int](intConsumer))) > > -->Requested Demand in the Producer *1* > > > I would expect that the "Request of demand" send to producers does not > depend on the steps of the flow? Or I'm a wrong. Moreover 4 seems somehow > strange to me. > Any help welcome. > > > Regards > > Wolfgang > Wolfgang > > -- > >>>>>>>>>> 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 http://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Typesafe - The software stack for applications that scale 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 akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
