Hi Victor, As far as I can see there is no way to guarantee that you will know if a message was discarded in the process step in any other way than writing your own custom stage.
You could for example create a stage that will record incoming messages before sending to process, and then also plug that in on the outgoing side of process (it would perhaps be a bidi-stage). This way you could keep track of elements sent into process but not coming out on the other side. If the messages have a sequence id that is only growing and doesn't have holes in it, detecting a missing message could be done with a custom stage that looks at the previous id and the current. Depending on use case there might be more strategies that could achieve what you want. -- Johan Akka Team On Wed, Aug 31, 2016 at 1:16 PM, Victor <[email protected]> wrote: > Hi, > > I've designed an AMQP graph where I need to ack or nack a message > depending on the processing result of the message: > > *AmqpSource* (consume) -> *Process* -> *Recover* -> *AmqpSink* (publish) > > All stages are under my control except the *Process* stage which is a > *Flow* and is implemented by a different user and is a black box from my > point of view. > > The *Recover* stage will emit an *ErrorMessage* or a *Message* depending > on the *Process* stage result, because I have to deal with *Process* > stage failures as I don't know what happen in this stage. > > The *AmqpSink* will publish the received message to an exchange and then > ack or nack the message depending if it's an *ErrorMessage* or a *Message* > . > > The problem I have is that in the *Process* stage the user may filter > incoming elements, so these elements will never arrive in the *AmqpSink* > and will never be acked. > > So to make it simple, if I have N elements out from the source, I MUST > have N elements in in the sink, whereas it's an *ErrorMessage* or a > *Message*, and all of that whatever happens in the *Process* stage. > Is there a way to catch dropped elements in a stream? How could I resolve > my problem? > > Thanks in advance, > Victor > > -- > >>>>>>>>>> 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. > -- Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
