Ticket #102: Propagate node state

Author: habarthierry-hue

src/RaftConsensus/FaasNet.RaftConsensus.Client/GossipClient.cs

@@ -43,6 +43,12 @@ public GossipClient(string url, int port) : this(new IPEndPoint(IPAddressHelper.
             return Send(package, cancellationToken: cancellationToken);
         }
 
+        public Task UpdateNodeState(string entityType, string entityId, string value, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var package = GossipPackageRequestBuilder.UpdateNodeState(string.Empty, 0, entityType, entityId, value);
+            return Send(package, cancellationToken: cancellationToken);
+        }
+
         public async Task Send(GossipPackage gossipPackage, int? timeoutMS = null, CancellationToken cancellationToken = default(CancellationToken))
         {
             var writeCtx = new WriteBufferContext();

src/RaftConsensus/FaasNet.RaftConsensus.Client/Messages/Gossip/GossipPackageRequestBuilder.cs

@@ -40,5 +40,16 @@ public static GossipPackage UpdateClusterNodes(string url, int port, ICollection
                 Nodes = clusterNodes
             };
         }
+
+        public static GossipPackage UpdateNodeState(string url, int port, string entityType, string entityId, string value)
+        {
+            return new GossipUpdateNodeStateRequest
+            {
+                Header = new GossipHeader(GossipCommands.UPDATE_NODE_STATE_REQUEST, url, port),
+                EntityId = entityId,
+                EntityType = entityType,
+                Value = value
+            };
+        }
     }
 }

src/RaftConsensus/FaasNet.RaftConsensus.Core/BaseNodeHost.cs

@@ -247,6 +247,8 @@ private async Task<GossipPackage> HandleGossipRequest(GossipHeartbeatResult requ
             var states = await _nodeStateStore.GetAllLastEntityTypes(TokenSource.Token);
             var stateToBeSynced = states.Where(s => request.States.Any(st => st.EntityType == s.EntityType && s.EntityVersion < st.EntityVersion))
                 .ToDictionary(s => s.EntityType, s => s.EntityVersion + 1);
+            var missingStates = request.States.Where(rs => !states.Any(s => s.EntityType == rs.EntityType));
+            foreach (var missingState in missingStates) stateToBeSynced.Add(missingState.EntityType, missingState.EntityVersion);
             if (!stateToBeSynced.Any()) return null;
             return GossipPackageRequestBuilder.Sync(_options.Url, _options.Port, stateToBeSynced);
         }

src/RaftConsensus/FaasNet.RaftConsensus.Core/Models/NodeState.cs

@@ -1,5 +1,8 @@
-namespace FaasNet.RaftConsensus.Core.Models
+using System.Diagnostics;
+
+namespace FaasNet.RaftConsensus.Core.Models
 {
+    [DebuggerDisplay("EntityType = {EntityType}, EntityId = {EntityId}, Value = {Value}")]
     public class NodeState
     {
         public string EntityType { get; set; }

tests/FaasNet.RaftConsensus.Tests/PeerHostFixture.cs

@@ -70,6 +70,32 @@ public async Task When_NodeIsStopped_Then_NodeBecomeUnreachable()
             Assert.Equal(4001, clusterNodes.First().Node.Port);
         }
 
+        [Fact]
+        public async Task When_StateIsAdded_Then_StorageIsUpdated()
+        {
+            // ARRANGE
+            var seedNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4000, new ConcurrentBag<ClusterNode>(), true);
+            var firstNode = BuildNodeHost(new ConcurrentBag<PeerInfo>(), 4001, new ConcurrentBag<ClusterNode>());
+            await seedNode.Start(CancellationToken.None);
+            await firstNode.Start(CancellationToken.None);
+            WaitNodeIsStarted(seedNode);
+            WaitNodeIsStarted(firstNode);
+            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.JoinNode("localhost", 4001);
+            var seedNodeStates = await WaitEntityTypes(seedNode, (nodes) => nodes.Count() == 2);
+            var firstNodeStates = await WaitEntityTypes(firstNode, (nodes) => nodes.Count() == 2);
+
+            // ACT
+            using (var gossipClient = new GossipClient("localhost", 4000)) await gossipClient.UpdateNodeState("Client", "id", "value");
+            var seedClient = (await WaitEntityTypes(seedNode, (nodes) => nodes.Any(n => n.EntityType == "Client"))).First(c => c.EntityType == "Client");
+            var firstNodeClient = (await WaitEntityTypes(firstNode, (nodes) => nodes.Any(n => n.EntityType == "Client"))).First(c => c.EntityType == "Client");
+
+            // ASSERT
+            Assert.NotNull(seedClient);
+            Assert.NotNull(firstNodeClient);
+            Assert.Equal("Client", seedClient.EntityType);
+            Assert.Equal("Client", firstNodeClient.EntityType);
+        }
+
         #endregion
 
         [Fact]