Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -53,23 +53,23 @@ protected AsyncReplayTimeoutException(SerializationInfo info, StreamingContext c
}

/// <summary>
/// TBD
/// Message that supplies the actor used by the proxy as its journal store.
/// </summary>
[Serializable]
public sealed class SetStore
{
/// <summary>
/// TBD
/// Creates a message containing the target journal-store actor.
/// </summary>
/// <param name="store">TBD</param>
/// <param name="store">The actor that handles journal protocol messages.</param>
/// <exception cref="ArgumentNullException">
/// This exception is thrown when the specified <paramref name="store"/> is undefined.
/// </exception>
public SetStore(IActorRef store) =>
Store = store ?? throw new ArgumentNullException(nameof(store), "SetStore requires non-null reference to store actor");

/// <summary>
/// TBD
/// The journal-store actor supplied to the proxy.
/// </summary>
public readonly IActorRef Store;
}
Expand All @@ -92,7 +92,7 @@ private InitTimeout() { }
private IActorRef _store;

/// <summary>
/// TBD
/// Initializes the journal proxy and prepares a timeout while it waits for its store actor.
/// </summary>
protected AsyncWriteProxyEx()
{
Expand All @@ -102,12 +102,12 @@ protected AsyncWriteProxyEx()
}

/// <summary>
/// TBD
/// The maximum wait before the proxy stops stashing messages for store initialization.
/// </summary>
public abstract TimeSpan Timeout { get; }

/// <summary>
/// TBD
/// Starts the initialization timeout before the journal's base startup logic runs.
/// </summary>
public override void AroundPreStart()
{
Expand All @@ -116,11 +116,11 @@ public override void AroundPreStart()
}

/// <summary>
/// TBD
/// Stashes messages until a store is supplied or the initialization timeout expires, then delegates normal handling.
/// </summary>
/// <param name="receive">TBD</param>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="receive">The journal's receive handler.</param>
/// <param name="message">The incoming message to process or stash.</param>
/// <returns><c>true</c> when initialization handling consumes the message; otherwise, the base journal result.</returns>
protected internal override bool AroundReceive(Receive receive, object message)
{
if (_isInitialized)
Expand Down Expand Up @@ -198,18 +198,18 @@ protected override Task DeleteMessagesToAsync(string persistenceId, long toSeque
}

/// <summary>
/// TBD
/// Replays messages from the configured store through a mediator that invokes the recovery callback and completes the replay task.
/// </summary>
/// <param name="context">TBD</param>
/// <param name="persistenceId">TBD</param>
/// <param name="fromSequenceNr">TBD</param>
/// <param name="toSequenceNr">TBD</param>
/// <param name="max">TBD</param>
/// <param name="recoveryCallback">TBD</param>
/// <param name="context">The actor context used to create a local replay mediator.</param>
/// <param name="persistenceId">The persistence identifier being recovered.</param>
/// <param name="fromSequenceNr">The first sequence number to replay.</param>
/// <param name="toSequenceNr">The last sequence number to replay.</param>
/// <param name="max">The maximum number of messages to replay.</param>
/// <param name="recoveryCallback">The callback invoked for each replayed persistent representation.</param>
/// <exception cref="TimeoutException">
/// This exception is thrown when the store has not been initialized.
/// </exception>
/// <returns>TBD</returns>
/// <returns>A task that completes when replay succeeds and faults when the store reports failure or the local inactivity timeout elapses.</returns>
public override Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> recoveryCallback)
{
if (_store == null)
Expand Down Expand Up @@ -253,15 +253,15 @@ private Task<T> StoreNotInitialized<T>()
}

/// <summary>
/// TBD
/// The stash used while waiting for the store actor to be set.
/// </summary>
public IStash Stash { get; set; }

public ITimerScheduler Timers { get; set; }
}

/// <summary>
/// TBD
/// Receives replay messages from the journal store, invokes the recovery callback, and completes or times out the replay request.
/// </summary>
internal class ReplayMediator : ActorBase
{
Expand All @@ -270,11 +270,11 @@ internal class ReplayMediator : ActorBase
private readonly TimeSpan _replayTimeout;

/// <summary>
/// TBD
/// Creates a mediator for one journal replay request.
/// </summary>
/// <param name="replayCallback">TBD</param>
/// <param name="replayCompletionPromise">TBD</param>
/// <param name="replayTimeout">TBD</param>
/// <param name="replayCallback">The callback invoked for each replayed persistent representation.</param>
/// <param name="replayCompletionPromise">The task completion source completed when replay succeeds or fails.</param>
/// <param name="replayTimeout">The inactivity timeout for receiving replay messages.</param>
public ReplayMediator(Action<IPersistentRepresentation> replayCallback, TaskCompletionSource<object> replayCompletionPromise, TimeSpan replayTimeout)
{
_replayCallback = replayCallback;
Expand All @@ -285,13 +285,13 @@ public ReplayMediator(Action<IPersistentRepresentation> replayCallback, TaskComp
}

/// <summary>
/// TBD
/// Processes replayed records, completion or failure notifications, and replay inactivity timeouts.
/// </summary>
/// <param name="message">TBD</param>
/// <param name="message">A replay protocol message or receive-timeout notification.</param>
/// <exception cref="AsyncReplayTimeoutException">
/// This exception is thrown when the replay timed out due to inactivity.
/// </exception>
/// <returns>TBD</returns>
/// <returns><c>true</c> for replay protocol messages and timeouts handled here; otherwise, <c>false</c>.</returns>
protected override bool Receive(object message)
{
switch (message)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ private InitTimeout() { }
private IActorRef _store;

/// <summary>
/// TBD
/// Initializes the snapshot-store proxy in an uninitialized state with no target store actor.
/// </summary>
protected SnapshotStoreProxy()
{
Expand All @@ -42,19 +42,19 @@ protected SnapshotStoreProxy()
}

/// <summary>
/// TBD
/// The maximum wait before the proxy stops stashing messages for store initialization.
/// </summary>
public abstract TimeSpan Timeout { get; }

/// <summary>
/// TBD
/// The stash used while waiting for the store actor to be set.
/// </summary>
public IStash Stash { get; set; }

public ITimerScheduler Timers { get; set; }

/// <summary>
/// TBD
/// Starts the initialization timeout before the snapshot store's base startup logic runs.
/// </summary>
public override void AroundPreStart()
{
Expand All @@ -63,11 +63,11 @@ public override void AroundPreStart()
}

/// <summary>
/// TBD
/// Stashes messages until a store is supplied or initialization times out, then delegates normal snapshot-store handling.
/// </summary>
/// <param name="receive">TBD</param>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="receive">The snapshot store's receive handler.</param>
/// <param name="message">The incoming message to process or stash.</param>
/// <returns><c>true</c> when initialization handling consumes the message; otherwise, the base snapshot-store result.</returns>
protected internal override bool AroundReceive(Receive receive, object message)
{
if (_isInitialized)
Expand Down
36 changes: 18 additions & 18 deletions src/contrib/cluster/Akka.Cluster.Sharding/ClusterSharding.cs
Original file line number Diff line number Diff line change
Expand Up @@ -129,16 +129,16 @@ public static HashCodeMessageExtractor Create(int maxNumberOfShards, Func<Msg, s
=> new Implementation(maxNumberOfShards, entityIdExtractor, messageExtractor);

/// <summary>
/// TBD
/// Gets the maximum shard count configured for this hash-based extractor.
/// </summary>
public readonly int MaxNumberOfShards;

private readonly Dictionary<int, string> _cachedIds;

/// <summary>
/// TBD
/// Initializes the cache of shard ids from the configured shard count.
/// </summary>
/// <param name="maxNumberOfShards">TBD</param>
/// <param name="maxNumberOfShards">The number of shard id strings to cache, numbered from zero.</param>
protected HashCodeMessageExtractor(int maxNumberOfShards)
{
MaxNumberOfShards = maxNumberOfShards;
Expand All @@ -150,29 +150,29 @@ protected HashCodeMessageExtractor(int maxNumberOfShards)
}

/// <summary>
/// TBD
/// Extracts the entity id from the incoming message.
/// </summary>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="message">The message received by the shard region.</param>
/// <returns>The entity id, or <see langword="null"/> when the message has no entity id.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public abstract EntityId? EntityId(Msg message);

/// <summary>
/// Default implementation pass on the message as is.
/// </summary>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="message">The message received by the shard region.</param>
/// <returns>The message to deliver to the entity; the default implementation returns <paramref name="message"/>.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public virtual Msg? EntityMessage(Msg message)
{
return message;
}

/// <summary>
/// TBD
/// Computes a shard id from the entity id in the message using this extractor's hash partitioning.
/// </summary>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="message">The message from which to extract an entity id.</param>
/// <returns>A shard id, or <see langword="null"/> when the extracted entity id is null or empty.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
[Obsolete("Use ShardId(string, object?) instead. Since v1.5.15")]
public virtual ShardId? ShardId(Msg message)
Expand Down Expand Up @@ -373,7 +373,7 @@ public ClusterSharding(ExtendedActorSystem system)
/// <summary>
/// Default HOCON settings for cluster sharding.
/// </summary>
/// <returns>TBD</returns>
/// <returns>The default cluster-sharding configuration with distributed-data defaults as fallback.</returns>
public static Config DefaultConfig()
{
return ConfigurationFactory.FromResource<ClusterSharding>("Akka.Cluster.Sharding.reference.conf")
Expand Down Expand Up @@ -1422,11 +1422,11 @@ public async Task<IActorRef> StartProxyAsync(string typeName, string role, IMess
/// <see cref="ClusterShardingGuardian.StartProxy"/> method before it
/// can be used here. Messages to the entity is always sent via the <see cref="Sharding.ShardRegion"/>.
/// </summary>
/// <param name="typeName">TBD</param>
/// <param name="typeName">The entity type name registered with sharding.</param>
/// <exception cref="ArgumentException">
/// Thrown when shard region for provided <paramref name="typeName"/> has not been started yet.
/// </exception>
/// <returns>TBD</returns>
/// <returns>The registered local shard region or proxy for the entity type.</returns>
public IActorRef ShardRegion(string typeName)
#pragma warning restore CS0419 // Ambiguous reference in cref attribute
{
Expand Down Expand Up @@ -1518,8 +1518,8 @@ public interface IMessageExtractor
/// If <see langword="null"/> is returned the message will be `unhandled`, i.e. posted as `Unhandled`
/// messages on the event stream
/// </summary>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="message">The incoming message to inspect.</param>
/// <returns>The entity id extracted from the message, or <see langword="null"/> when the message is not handled.</returns>
EntityId? EntityId(Msg message);

/// <summary>
Expand All @@ -1528,8 +1528,8 @@ public interface IMessageExtractor
/// message to support wrapping in message envelope that is unwrapped before
/// sending to the entity actor.
/// </summary>
/// <param name="message">TBD</param>
/// <returns>TBD</returns>
/// <param name="message">The incoming message to inspect.</param>
/// <returns>The message to send to the entity, which may be a wrapper or the original message.</returns>
Msg? EntityMessage(Msg message);

/// <summary>
Expand Down
Loading
Loading