Class AtLeastOnceDeliveryActor

Persistent actor type that sends messages with at-least-once delivery semantics to destinations. It takes care of re-sending messages when they haven't been confirmed withing expected timeout. Use the Deliver(ActorPath, Func<Int64, Object>, Boolean) method to send a message to a destination. Call the ConfirmDelivery(Int64) method when destination has replied with a confirmation message.

At-least-once delivery implies that the 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 with RedeliverInterval. After a number of delivery attempts a UnconfirmedWarning message will be sent to Self. The re-sending will continue, but you may choose ConfirmDelivery(Int64) to cancel re-sending.

This actor type has a state consisting of unconfirmed messages and a sequence number. It does not store this state itself. You must persist events corresponding to the Deliver(ActorPath, Func<Int64, Object>, Boolean) and ConfirmDelivery(Int64) invocations from your PersistentActor so that the state can be restored by calling the same methods during the recovery phase of the PersistentActor. Sometimes these events can be derived from other business level events, and sometimes you must create separate events. During recovery calls to Deliver(ActorPath, Func<Int64, Object>, Boolean) will not send out the message, but it will be sent later if no matching ConfirmDelivery(Int64) was performed.

Support for snapshot is provided by GetDeliverySnapshot() and SetDeliverySnapshot(AtLeastOnceDeliverySnapshot). 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.

Inheritance

Object

ActorBase

Eventsourced

PersistentActor

AtLeastOnceDeliveryActor

Implements

IInternalActor

IPersistentIdentity

IPersistenceStash

IWithUnboundedStash

IWithUnrestrictedStash

IActorStash

IRequiresMessageQueue<IUnboundedDequeBasedMessageQueueSemantics>

IPersistenceRecovery

Inherited Members

PersistentActor.Receive(Object)

Eventsourced.Extension

Eventsourced.Log

Eventsourced.PersistenceId

Eventsourced.Recovery

Eventsourced.InternalStashOverflowStrategy

Eventsourced.Stash

Eventsourced.JournalPluginId

Eventsourced.SnapshotPluginId

Eventsourced.Journal

Eventsourced.SnapshotStore

Eventsourced.SnapshotterId

Eventsourced.IsRecovering

Eventsourced.IsRecoveryFinished

Eventsourced.LastSequenceNr

Eventsourced.SnapshotSequenceNr

Eventsourced.LoadSnapshot(String, SnapshotSelectionCriteria, Int64)

Eventsourced.SaveSnapshot(Object)

Eventsourced.DeleteSnapshot(Int64)

Eventsourced.DeleteSnapshots(SnapshotSelectionCriteria)

Eventsourced.ReceiveRecover(Object)

Eventsourced.ReceiveCommand(Object)

Eventsourced.Persist<TEvent>(TEvent, Action<TEvent>)

Eventsourced.PersistAll<TEvent>(IEnumerable<TEvent>, Action<TEvent>)

Eventsourced.PersistAsync<TEvent>(TEvent, Action<TEvent>)

Eventsourced.PersistAllAsync<TEvent>(IEnumerable<TEvent>, Action<TEvent>)

Eventsourced.DeferAsync<TEvent>(TEvent, Action<TEvent>)

Eventsourced.DeleteMessages(Int64)

Eventsourced.OnRecoveryFailure(Exception, Object)

Eventsourced.OnPersistFailure(Exception, Object, Int64)

Eventsourced.OnPersistRejected(Exception, Object, Int64)

Eventsourced.RunTask(Func<Task>)

Eventsourced.UnstashFilterPredicate

Eventsourced.AroundPreStart()

Eventsourced.AroundPostRestart(Exception, Object)

Eventsourced.Unhandled(Object)

ActorBase.Sender

ActorBase.Self

ActorBase.IInternalActor.ActorContext

ActorBase.Context

ActorBase.EmptyReceive

ActorBase.Become(Receive)

ActorBase.BecomeStacked(Receive)

ActorBase.UnbecomeStacked()

ActorBase.SetReceiveTimeout(Nullable<TimeSpan>)

ActorBase.PreStart()

ActorBase.PreRestart(Exception, Object)

ActorBase.PostRestart(Exception)

ActorBase.PostStop()

ActorBase.SupervisorStrategy()

Object.Equals(Object)

Object.Equals(Object, Object)

Object.GetHashCode()

Object.GetType()

Object.MemberwiseClone()

Object.ReferenceEquals(Object, Object)

Object.ToString()

Namespace: Akka.Persistence

Assembly: Akka.Persistence.dll

Syntax

public abstract class AtLeastOnceDeliveryActor : PersistentActor, IInternalActor, IPersistentIdentity, IPersistenceStash, IWithUnboundedStash, IWithUnrestrictedStash, IActorStash, IRequiresMessageQueue<IUnboundedDequeBasedMessageQueueSemantics>, IPersistenceRecovery

Constructors

AtLeastOnceDeliveryActor()

Initializes a new instance of the AtLeastOnceDeliveryActor class.

Declaration

protected AtLeastOnceDeliveryActor()

AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings)

Initializes a new instance of the AtLeastOnceDeliveryActor class.

Declaration

protected AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings settings)

Parameters

Type	Name	Description
PersistenceSettings.AtLeastOnceDeliverySettings	settings	TBD

AtLeastOnceDeliveryActor(Func<PersistenceSettings.AtLeastOnceDeliverySettings, PersistenceSettings.AtLeastOnceDeliverySettings>)

Initializes a new instance of the AtLeastOnceDeliveryActor class.

Declaration

protected AtLeastOnceDeliveryActor(Func<PersistenceSettings.AtLeastOnceDeliverySettings, PersistenceSettings.AtLeastOnceDeliverySettings> overrideSettings)

Parameters

Type	Name	Description
Func<PersistenceSettings.AtLeastOnceDeliverySettings, PersistenceSettings.AtLeastOnceDeliverySettings>	overrideSettings	A lambda to tweak the default AtLeastOnceDelivery settings.

Properties

MaxUnconfirmedMessages

Maximum number of unconfirmed messages, that this actor is allowed to hold in the memory. if this number is exceeded, Deliver(ActorPath, Func<Int64, Object>, Boolean) will not accept more messages and it will throw MaxUnconfirmedMessagesExceededException.

The default value can be configure with the 'akka.persistence.at-least-once-delivery.max-unconfirmed-messages' configuration key. Custom value may be provided via the AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings) constructor.

Declaration

public int MaxUnconfirmedMessages { get; }

Property Value

Type	Description
Int32

RedeliverInterval

Interval between redelivery attempts.

The default value can be configure with the 'akka.persistence.at-least-once-delivery.redeliver-interval' configuration key. Custom value may be provided via the AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings) constructor.

Declaration

public TimeSpan RedeliverInterval { get; }

Property Value

Type	Description
TimeSpan

RedeliveryBurstLimit

Maximum number of unconfirmed messages that will be sent at each redelivery burst (burst frequency is half of the redelivery interval). If there's a lot of unconfirmed messages (e.g. if the destination is not available for a long time), this helps prevent an overwhelming amount of messages to be sent at once.

The default value can be configure with the 'akka.persistence.at-least-once-delivery.redelivery-burst-limit' configuration key. Custom value may be provided via the AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings) constructor.

Declaration

public int RedeliveryBurstLimit { get; }

Property Value

Type	Description
Int32

UnconfirmedCount

Number of messages that have not been confirmed yet.

Declaration

public int UnconfirmedCount { get; }

Property Value

Type	Description
Int32

WarnAfterNumberOfUnconfirmedAttempts

After this number of delivery attempts a UnconfirmedWarning message will be sent to Self. The count is reset after restart.

The default value can be configure with the 'akka.persistence.at-least-once-delivery.warn-after-number-of-unconfirmed-attempts' configuration key. Custom value may be provided via the AtLeastOnceDeliveryActor(PersistenceSettings.AtLeastOnceDeliverySettings) constructor.

Declaration

public int WarnAfterNumberOfUnconfirmedAttempts { get; }

Property Value

Type	Description
Int32

Methods

AroundPostStop()

Can be overridden to intercept calls to PostStop. Calls PostStop by default..

Declaration

public override void AroundPostStop()

Overrides

Eventsourced.AroundPostStop()

AroundPreRestart(Exception, Object)

Can be overridden to intercept calls to PreRestart. Calls PreRestart by default.

Declaration

public override void AroundPreRestart(Exception cause, object message)

Parameters

Type	Name	Description
Exception	cause	The cause.
Object	message	The message.

Overrides

Eventsourced.AroundPreRestart(Exception, Object)

AroundReceive(Receive, Object)

TBD

Declaration

protected override bool AroundReceive(Receive receive, object message)

Parameters

Type	Name	Description
Receive	receive	TBD
Object	message	TBD

Returns

Type	Description
Boolean	TBD

Overrides

Eventsourced.AroundReceive(Receive, Object)

ConfirmDelivery(Int64)

Call this method when a message has been confirmed by the destination, or to abort re-sending.

Declaration

public bool ConfirmDelivery(long deliveryId)

Parameters

Type	Name	Description
Int64	deliveryId	TBD

Returns

Type	Description
Boolean	True the first time the deliveryId is confirmed, false for duplicate confirmations.

Deliver(ActorPath, Func<Int64, Object>)

Send the message created with deliveryMessageMapper function to the destination actor. It will retry sending the message until the delivery is confirmed with ConfirmDelivery(Int64). Correlation between these two methods is performed by deliveryId that is provided as parameter to the deliveryMessageMapper function. The deliveryId is typically passed in the message to the destination, which replies with a message containing the same 'deliveryId'.

The 'deliveryId' is a strictly monotonically increasing sequence number without gaps. The same sequence is used for all destinations of the actor, i.e. when sending to multiple destinations the destinations will see gaps in the sequence if no translation is performed.

During recovery this method will not send out the message, but it will be sent later if no matching ConfirmDelivery(Int64) was performed.

Declaration

public void Deliver(ActorPath destination, Func<long, object> deliveryMessageMapper)

Parameters

Type	Name	Description
ActorPath	destination	TBD
Func<Int64, Object>	deliveryMessageMapper	TBD

Exceptions

Type	Condition
MaxUnconfirmedMessagesExceededException	Thrown when UnconfirmedCount is greater than or equal to MaxUnconfirmedMessages.

Deliver(ActorSelection, Func<Int64, Object>)

Send the message created with deliveryMessageMapper function to the destination actor. It will retry sending the message until the delivery is confirmed with ConfirmDelivery(Int64). Correlation between these two methods is performed by deliveryId that is provided as parameter to the deliveryMessageMapper function. The deliveryId is typically passed in the message to the destination, which replies with a message containing the same 'deliveryId'.

The 'deliveryId' is a strictly monotonically increasing sequence number without gaps. The same sequence is used for all destinations of the actor, i.e. when sending to multiple destinations the destinations will see gaps in the sequence if no translation is performed.

During recovery this method will not send out the message, but it will be sent later if no matching ConfirmDelivery(Int64) was performed.

Declaration

public void Deliver(ActorSelection destination, Func<long, object> deliveryMessageMapper)

Parameters

Type	Name	Description
ActorSelection	destination	TBD
Func<Int64, Object>	deliveryMessageMapper	TBD

Exceptions

Type	Condition
MaxUnconfirmedMessagesExceededException	Thrown when UnconfirmedCount is greater than or equal to MaxUnconfirmedMessages.
NotSupportedException	TBD

GetDeliverySnapshot()

Full state of the AtLeastOnceDeliveryActor. It can be saved with SaveSnapshot(Object). During recovery the snapshot received in SnapshotOffer should be set with SetDeliverySnapshot(AtLeastOnceDeliverySnapshot).

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.

Declaration

public AtLeastOnceDeliverySnapshot GetDeliverySnapshot()

Returns

Type	Description
AtLeastOnceDeliverySnapshot	TBD

OnReplaySuccess()

Called whenever a message replay succeeds.

Declaration

protected override void OnReplaySuccess()

Overrides

Eventsourced.OnReplaySuccess()

SetDeliverySnapshot(AtLeastOnceDeliverySnapshot)

If snapshot from GetDeliverySnapshot() was saved it will be received during recovery phase in a SnapshotOffer message and should be set with this method.

Declaration

public void SetDeliverySnapshot(AtLeastOnceDeliverySnapshot snapshot)

Parameters

Type	Name	Description
AtLeastOnceDeliverySnapshot	snapshot	TBD

Implements

IInternalActor

IPersistentIdentity

IPersistenceStash

IWithUnboundedStash

IWithUnrestrictedStash

IActorStash

IRequiresMessageQueue<T>

IPersistenceRecovery

Extension Methods

ObjectExtensions.IsDefaultForType<T>(T)

ObjectExtensions.AsOption<T>(T)

Extensions.AsInstanceOf<T>(Object)