Skip to content

Commit

Permalink
make member expiration configurable
Browse files Browse the repository at this point in the history
  • Loading branch information
rogeralsing committed Sep 18, 2023
1 parent 24fa206 commit 25eeda0
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 7 deletions.
15 changes: 15 additions & 0 deletions src/Proto.Cluster/ClusterConfig.cs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@
using System.Linq;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using JetBrains.Annotations;
using Proto.Cluster.Gossip;
using Proto.Cluster.Identity;
using Proto.Cluster.PubSub;
using Proto.Remote;
Expand Down Expand Up @@ -382,4 +384,17 @@ public static ClusterConfig Setup(
IIdentityLookup identityLookup
) =>
new(clusterName, clusterProvider, identityLookup);

/// <summary>
/// The code to run when a member is expired from the cluster.
/// </summary>
public Func<Cluster, Task> BlockExpiredMembers { get; init; } = Gossiper.BlockExpiredMembers;

/// <summary>
/// Configures the code to run when a member is expired from the cluster.
/// </summary>
/// <param name="blockExpiredMembers"></param>
/// <returns></returns>
public ClusterConfig WithBlockExpiredMembers(Func<Cluster, Task> blockExpiredMembers) =>
this with { BlockExpiredMembers = blockExpiredMembers };
}
17 changes: 10 additions & 7 deletions src/Proto.Cluster/Gossip/Gossiper.cs
Original file line number Diff line number Diff line change
Expand Up @@ -270,18 +270,21 @@ private async Task BlockExpiredHeartbeats()
{
return;
}

await _cluster.Config.BlockExpiredMembers(_cluster);
}

var t = await GetStateEntry(GossipKeys.Heartbeat).ConfigureAwait(false);

var blockList = _cluster.System.Remote().BlockList;
public static async Task BlockExpiredMembers(Cluster cluster)
{
var gossipState = await cluster.Gossip. GetStateEntry(GossipKeys.Heartbeat).ConfigureAwait(false);
var blockList = cluster.Remote.BlockList;
var alreadyBlocked = blockList.BlockedMembers;

//new blocked members
var blocked = (from x in t
var blocked = (from x in gossipState
//never block ourselves
where x.Key != _cluster.System.Id
where x.Key != cluster.System.Id
//pick any entry that is too old
where x.Value.Age > _cluster.Config.HeartbeatExpiration
where x.Value.Age > cluster.Config.HeartbeatExpiration
//and not already part of the block list
where !alreadyBlocked.Contains(x.Key)
select x.Key)
Expand Down

0 comments on commit 25eeda0

Please sign in to comment.