raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Merge pull request #18196 from akka/wip-17033-AtLeastOnceDeliveryLike-patriknw

Browse source 
=per #17033 AtLeastOnceDelivery extends PersistentActor
This commit is contained in:
Konrad Malawski 2015-08-13 22:31:15 +02:00
commit d84492e581
1 changed files with 25 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

38
akka-persistence/src/main/scala/akka/persistence/AtLeastOnceDelivery.scala
View file

@ -14,9 +14,9 @@ object AtLeastOnceDelivery {
/**
* Snapshot of current `AtLeastOnceDelivery` state. Can be retrieved with
* [[AtLeastOnceDelivery#getDeliverySnapshot]] and saved with [[PersistentActor#saveSnapshot]].
* [[AtLeastOnceDeliveryLike#getDeliverySnapshot]] and saved with [[PersistentActor#saveSnapshot]].
* During recovery the snapshot received in [[SnapshotOffer]] should be set
* with [[AtLeastOnceDelivery#setDeliverySnapshot]].
* with [[AtLeastOnceDeliveryLike#setDeliverySnapshot]].
*/
@SerialVersionUID(1L)
case class AtLeastOnceDeliverySnapshot(currentDeliveryId: Long, unconfirmedDeliveries: immutable.Seq[UnconfirmedDelivery])
@ -33,7 +33,7 @@ object AtLeastOnceDelivery {
}
/**
* @see [[AtLeastOnceDelivery#warnAfterNumberOfUnconfirmedAttempts]]
* @see [[AtLeastOnceDeliveryLike#warnAfterNumberOfUnconfirmedAttempts]]
*/
@SerialVersionUID(1L)
case class UnconfirmedWarning(unconfirmedDeliveries: immutable.Seq[UnconfirmedDelivery]) {
@ -58,11 +58,14 @@ object AtLeastOnceDelivery {
}
/**
* @see [[AtLeastOnceDelivery#maxUnconfirmedMessages]]
* @see [[AtLeastOnceDeliveryLike#maxUnconfirmedMessages]]
*/
class MaxUnconfirmedMessagesExceededException(message: String) extends RuntimeException(message)
private object Internal {
/**
* INTERNAL API
*/
private[akka] object Internal {
case class Delivery(destination: ActorPath, message: Any, timestamp: Long, attempt: Int)
case object RedeliveryTick
}
@ -72,17 +75,17 @@ object AtLeastOnceDelivery {
/**
* Mix-in this trait with your `PersistentActor` to send messages with at-least-once
* delivery semantics to destinations. It takes care of re-sending messages when they
* have not been confirmed within a configurable timeout. Use the [[#deliver]] method to
* send a message to a destination. Call the [[#confirmDelivery]] method when the destination
* have not been confirmed within a configurable timeout. Use the [[AtLeastOnceDeliveryLike#deliver]] method to
* send a message to a destination. Call the [[AtLeastOnceDeliveryLike#confirmDelivery]] method when the destination
* has replied with a confirmation message.
*
* At-least-once delivery implies that original message send order is not always retained
* and the destination may receive duplicate messages due to possible resends.
*
* The interval between redelivery attempts can be defined by [[#redeliverInterval]].
* The interval between redelivery attempts can be defined by [[AtLeastOnceDeliveryLike#redeliverInterval]].
* After a number of delivery attempts a [[AtLeastOnceDelivery.UnconfirmedWarning]] message
* will be sent to `self`. The re-sending will still continue, but you can choose to call
* [[#confirmDelivery]] to cancel the re-sending.
* [[AtLeastOnceDeliveryLike#confirmDelivery]] to cancel the re-sending.
*
* The `AtLeastOnceDelivery` trait has a state consisting of unconfirmed messages and a
* sequence number. It does not store this state itself. You must persist events corresponding
@ -93,14 +96,21 @@ object AtLeastOnceDelivery {
* will not send out the message, but it will be sent later if no matching `confirmDelivery`
* was performed.
*
* Support for snapshots is provided by [[#getDeliverySnapshot]] and [[#setDeliverySnapshot]].
* Support for snapshots is provided by [[AtLeastOnceDeliveryLike#getDeliverySnapshot]] and [[AtLeastOnceDeliveryLike#setDeliverySnapshot]].
* The `AtLeastOnceDeliverySnapshot` contains the full delivery state, including unconfirmed messages.
* If you need a custom snapshot for other parts of the actor state you must also include the
* `AtLeastOnceDeliverySnapshot`. It is serialized using protobuf with the ordinary Akka
* serialization mechanism. It is easiest to include the bytes of the `AtLeastOnceDeliverySnapshot`
* as a blob in your custom snapshot.
*
* @see [[AtLeastOnceDeliveryLike]]
*/
trait AtLeastOnceDelivery extends Eventsourced {
trait AtLeastOnceDelivery extends PersistentActor with AtLeastOnceDeliveryLike
/**
* @see [[AtLeastOnceDelivery]]
*/
trait AtLeastOnceDeliveryLike extends Eventsourced {
import AtLeastOnceDelivery._
import AtLeastOnceDelivery.Internal._
@ -349,8 +359,9 @@ trait AtLeastOnceDelivery extends Eventsourced {
* Full documentation in [[AtLeastOnceDelivery]].
*
* @see [[AtLeastOnceDelivery]]
* @see [[AtLeastOnceDeliveryLike]]
*/
abstract class UntypedPersistentActorWithAtLeastOnceDelivery extends UntypedPersistentActor with AtLeastOnceDelivery {
abstract class UntypedPersistentActorWithAtLeastOnceDelivery extends UntypedPersistentActor with AtLeastOnceDeliveryLike {
/**
* Java API: Send the message created by the `deliveryIdToMessage` function to
* the `destination` actor. It will retry sending the message until
@ -406,8 +417,9 @@ abstract class UntypedPersistentActorWithAtLeastOnceDelivery extends UntypedPers
* Full documentation in [[AtLeastOnceDelivery]].
*
* @see [[AtLeastOnceDelivery]]
* @see [[AtLeastOnceDeliveryLike]]
*/
abstract class AbstractPersistentActorWithAtLeastOnceDelivery extends AbstractPersistentActor with AtLeastOnceDelivery {
abstract class AbstractPersistentActorWithAtLeastOnceDelivery extends AbstractPersistentActor with AtLeastOnceDeliveryLike {
/**
* Java API: Send the message created by the `deliveryIdToMessage` function to
* the `destination` actor. It will retry sending the message until