It looks like you don't use the deliveryId in the Msg, and therefore the `confirmDelivery` will not correlate with the `deliver` and it will continue resending.
Also, remove or change some settings from the config. These don't make much sense: redelivery-burst-limit = 0 max-unconfirmed-messages = 0 /Patrik On Mon, Dec 29, 2014 at 2:39 AM, shashank Jain <[email protected]> wrote: > Hi > I have created this data pipeline. > Sending 100 messages to persistent actor. Code below. Also attaching > source zipped. > > public class MyPersistentActor extends > UntypedPersistentActorWithAtLeastOnceDelivery > { > private final ActorPath destination; > static int count; > @Override > public String persistenceId() { return "persistence-id"; } > > public MyPersistentActor(ActorPath destination) { > this.destination = destination; > } > > @Override > public void onReceiveCommand(Object message) { > > > if (message instanceof String) { > > String s = (String) message; > persist(new MsgSent(s), new Procedure<MsgSent>() { > public void apply(MsgSent evt) { > updateState(evt); > } > }); > } else if (message instanceof Confirm) { > > > Confirm confirm = (Confirm) message; > // > persist(new MsgConfirmed(confirm.deliveryId), new > Procedure<MsgConfirmed>() { > public void apply(MsgConfirmed evt) { > //System.out.println("confirmation message for message > id="+evt.deliveryId); > updateState(evt); > } > }); > } else { > unhandled(message); > } > } > > @Override > public void onReceiveRecover(Object event) { > updateState(event); > } > > void updateState(Object event) { > if (event instanceof MsgSent) { > Random randomno = new Random(); > > // get next long value > final long id = randomno.nextLong(); > final MsgSent evt = (MsgSent) event; > > deliver(*destination*, new Function<Long, Object>() { > public Object apply(Long deliveryId) { > System.out.println("message sent "+id); > ///getContext().setReceiveTimeout(Duration.create("3 second")); > > return new Msg(id, evt.s); > } > }); > } else if (event instanceof MsgConfirmed) { > > > final MsgConfirmed evt = (MsgConfirmed) event; > System.out.println("message before confirmed "+evt.deliveryId); > boolean val= confirmDelivery(evt.deliveryId); > > // System.out.println("message confirmed "+evt.deliveryId+" with > boolean="+val); > } > } > > In deliver method we call an actor which is an http client making http > calls. > > Code snippet > > if(message instanceof Msg) > { > //getContext().setReceiveTimeout(Duration.create("1 millisecond")); > //turn off receive timeout > Msg msg=(Msg)message; > long start=System.nanoTime(); > > *int statusCode=simpleGet();* > //int statusCode=200; > long end=System.nanoTime(); > > // if(statusCode==200) > // { > getSender().tell(new Confirm(msg.deliveryId), getSelf()); > > } > > Instead of 100 calls into this method, I get sometimes 150, sometimes 200 > but never exact 100. I understand that this is setup for atleast once > delivery and the http call might have taken time due to which the message > is resent(typical to atleast once mode). > > Question is how do I control this behaviour. Say how can I increase this > timeout so that message is only resent after the configured time. > > Also this is the application.conf file > > > > akka { > logger-startup-timeout = 25s > typed { > # Default timeout for typed actor methods with non-void return type > timeout = 15s > } > > persistence { > > journal { > > # Maximum size of a persistent message batch written to the journal. > max-message-batch-size = 200 > } > at-least-once-delivery { > # Interval between redelivery attempts > * redeliver-interval = 15s* > > # Maximum number of unconfirmed messages that will be sent in one > redelivery burst > redelivery-burst-limit = 0 > > # After this number of delivery attempts a `ReliableRedelivery. > UnconfirmedWarning` > # message will be sent to the actor. > warn-after-number-of-unconfirmed-attempts = 1 > > # Maximum number of unconfirmed messages that an actor with > AtLeastOnceDelivery is > # allowed to hold in memory. > max-unconfirmed-messages = 0 > } > > } > > > > > akka.actor.deployment { > /myrouter1 { > router = round-robin > nr-of-instances = 5 > } > } > > } > > -- > >>>>>>>>>> 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. > -- Patrik Nordwall Typesafe <http://typesafe.com/> - Reactive apps on the JVM Twitter: @patriknw -- >>>>>>>>>> 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.
