using System.Collections.Immutable; using System.Net; using System.Net.Sockets; using Akka.Actor; using Akka.Cluster; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Shouldly; using Xunit; namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests; /// /// Cluster-002: must drop a member the moment it /// is leaving / exiting / downed, not several seconds later at MemberRemoved. The cluster /// events are fed straight into HandleMemberEvent, so no gossip timing is involved. /// public sealed class ClusterRoleInfoMemberEventTests { private const string SystemName = "clusterroleinfo-member-event-test"; private static int FreePort() { var listener = new TcpListener(IPAddress.Loopback, 0); listener.Start(); var port = ((IPEndPoint)listener.LocalEndpoint).Port; listener.Stop(); return port; } // Member's factory is internal to Akka.Cluster; reflection is the only way to mint one without // forming a real cluster (the whole point here is to avoid gossip timing). private static Member NewMember(int port, MemberStatus status) { // Pick the 5-argument overload: (UniqueAddress, upNumber, status, roles, appVersion). var create = typeof(Member) .GetMethods(System.Reflection.BindingFlags.Static | System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Public) .Single(m => m.Name == "Create" && m.GetParameters().Length == 5); var versionType = create.GetParameters()[4].ParameterType; var args = new object?[] { new UniqueAddress(new Address("akka.tcp", SystemName, "127.0.0.1", port), 1), 1, status, ImmutableHashSet.Create("driver"), versionType.IsValueType ? Activator.CreateInstance(versionType) : null, }; return (Member)create.Invoke(null, args)!; } private static async Task WithInfoAsync(Func body) { var port = FreePort(); var options = new AkkaClusterOptions { SystemName = SystemName, Hostname = "127.0.0.1", PublicHostname = "127.0.0.1", Port = port, Roles = new[] { "driver" }, SeedNodes = Array.Empty(), }; var system = ActorSystem.Create(SystemName, Akka.Configuration.ConfigurationFactory.ParseString( "akka.actor.provider = cluster\nakka.remote.dot-netty.tcp.hostname = 127.0.0.1\n" + $"akka.remote.dot-netty.tcp.port = {port}")); try { using var info = new ClusterRoleInfo(system, Options.Create(options), NullLogger.Instance); await body(info); } finally { await system.Terminate(); } } /// Each departing transition removes the member at once; the Up add is the positive control. [Theory] [InlineData("left")] [InlineData("exited")] [InlineData("downed")] [InlineData("removed")] public async Task Departing_transition_removes_the_member_immediately(string transition) { await WithInfoAsync(info => { var up = NewMember(11111, MemberStatus.Up); info.HandleMemberEvent(new ClusterEvent.MemberUp(up)); info.MembersWithRole("driver").Count.ShouldBe(1, "positive control: the Up member is listed"); ClusterEvent.IMemberEvent evt = transition switch { "left" => new ClusterEvent.MemberLeft(NewMember(11111, MemberStatus.Leaving)), "exited" => new ClusterEvent.MemberExited(NewMember(11111, MemberStatus.Exiting)), "downed" => new ClusterEvent.MemberDowned(NewMember(11111, MemberStatus.Down)), _ => new ClusterEvent.MemberRemoved(NewMember(11111, MemberStatus.Removed), MemberStatus.Exiting), }; info.HandleMemberEvent(evt); info.MembersWithRole("driver").ShouldBeEmpty(); return Task.CompletedTask; }); } }