:) I try to extract a sample from my code which shows what I'm doing. At least when I call OnNext twice I'll get the following exception
*java.lang.IllegalStateException: onNext is not allowed when the stream has not requested elements, totalDemand was 0* this exception I do understand. Am Freitag, 20. Februar 2015 09:29:49 UTC+1 schrieb drewhk: > > Hi Wolfgang, > > On Thu, Feb 19, 2015 at 7:35 PM, Wolfgang Friedl <[email protected] > <javascript:>> wrote: > >> Hi Endre >> >> My publisher is quite simple! It is an actor (extending the >> ActorPublisher) waiting for incoming messages. He puts the messages in the >> buffer if the totaldemand is > 0 I call onNext. >> If I would call onNext with totaldemand == 0 I would anyway get an >> Exception! But this is not the case. >> >> >> My sink is also an Actor (extending the ActorSubscriber) using the >> WatermarkRequestStrategy. I do not know what I can do wrong here. >> > > Me neither :) Can you put together a small code snippet that is at least > similar to what you are doing? I would like to figure out what is wrong. > > -Endre > > > >> >> Regards >> >> Wolfgang >> >> >> >> Am Donnerstag, 19. Februar 2015 16:53:44 UTC+1 schrieb drewhk: >>> >>> Hi Wolfgang, >>> >>> >>> >>> On Thu, Feb 19, 2015 at 4:47 PM, Wolfgang Friedl <[email protected] >>> > wrote: >>> >>>> There is one thing which came int to my mind. >>>> >>>> We are using mapConcat which does have the signature (f: Out ⇒ >>>> immutable.Seq[T]) >>>> >>>> Could it be that in a case were more elements are published the buffer >>>> of the next processor gets an overflow? >>>> >>> >>> No, this exception results from the violation of the Reactive Streams >>> protocol. While mapConcat might have a bug, in this case I suspect more >>> your publisher source and subscriber sink. What do they do? If they issue >>> more onNext than they were requested that will overrun the buffer since >>> they are not allowed to do that. >>> >>> >>> >>>> I guess in that case it would help to increase the inputbuffer size in >>>> the materializerSettings? >>>> >>> >>> That will not work. The buffer overrun occures not because the buffer is >>> small but because the producer does not respect the protocol. >>> >>> -Endre >>> >>> >>>> >>>> Regards >>>> >>>> Wolfgang >>>> >>>> >>>> Am Donnerstag, 19. Februar 2015 15:52:07 UTC+1 schrieb Wolfgang Friedl: >>>> >>>>> Hi Endre! >>>>> >>>>> Well thats not so easy. We use the flow internally to bulild a >>>>> "pipline" of processing steps (we recursively adding our own processables >>>>> and tighten them togehter by using map and mapConcat). So its difficult >>>>> to >>>>> add this peace of code. >>>>> >>>>> Finnaly we simply start this stream by calling >>>>> >>>>> _flow.runWith(PublisherSource(producer), SubscriberSink(consumer)) >>>>> >>>>> We used this approach since 3 month and today it was the first that I >>>>> observed that kind of exception on one of our systems. Currently we are >>>>> using the version 1.0-M2. >>>>> >>>>> We use the default MaterializerSettings (no changes at the Inputbuffer >>>>> size). As mentioned the producer is a simple Actor pushing message into >>>>> the >>>>> flow if requested. >>>>> >>>>> >>>>> >>>>> Regards >>>>> >>>>> Wolfgang >>>>> >>>>> >>>>> >>>>> >>>>> >>>>> >>>>> Am Donnerstag, 19. Februar 2015 15:28:52 UTC+1 schrieb drewhk: >>>>>> >>>>>> Hi Wolfgang, >>>>>> >>>>>> On Thu, Feb 19, 2015 at 3:24 PM, Wolfgang Friedl < >>>>>> [email protected]> wrote: >>>>>> >>>>>>> Hi togehter! >>>>>>> >>>>>>> I observed the same exception ( Input buffer overrun) on my side >>>>>>> today. Even so I read this task I'm not sure if I got it right. >>>>>>> >>>>>>> I do have a Flow were I put some "processing steps" together with >>>>>>> map and mapConcat. I've written the producer for the Flow on my own. >>>>>>> The Producer simply calls onNext() when a demand is requested. Due >>>>>>> to the implementation of akka.stream a call of OnNext without an demand >>>>>>> will end up in an exception. So how is it possible to get an internal >>>>>>> buffer overflow? >>>>>>> >>>>>> >>>>>> Without seeing some code that reproduces the issue it is hard to say >>>>>> anything. Can you provide us a simple snippet that triggers the failure? >>>>>> >>>>>> -Endre >>>>>> >>>>>> >>>>>>> >>>>>>> I've to say that I today observed the exception the first time even >>>>>>> so we are using the same kind of implementation for a while (2 month) >>>>>>> >>>>>>> Regards >>>>>>> >>>>>>> Wolfgang >>>>>>> >>>>>>> >>>>>>> >>>>>>> >>>>>>> >>>>>>> Am Montag, 16. Februar 2015 14:07:51 UTC+1 schrieb drewhk: >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> On Mon, Feb 16, 2015 at 1:24 PM, Carl Pulley < >>>>>>>> [email protected]> wrote: >>>>>>>> >>>>>>>>> Thanks for that (and the AutoPublisher hint). It was a great help >>>>>>>>> in seeing what I was misunderstanding! >>>>>>>>> >>>>>>>> >>>>>>>> Let's hope so :) If it does not work, then we have a bug, but >>>>>>>> thankfully we also will have a reproducer ;) >>>>>>>> >>>>>>>> -Endre >>>>>>>> >>>>>>>> >>>>>>>>> >>>>>>>>> Carl. >>>>>>>>> >>>>>>>>> -- >>>>>>>>> >>>>>>>>>> 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 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/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 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]. >>>> 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. >> > > -- >>>>>>>>>> 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.
