DVLP-013: Add gossip protocol

src/main/java/org/cache/Main.java

@@ -1,7 +1,11 @@
 package org.cache;
 
 import org.cache.cluster.CacheNode;
+import org.cache.cluster.ClusterGossipService;
 import org.cache.cluster.ClusterHealthMonitor;
+import org.cache.cluster.ClusterMembership;
+import org.cache.cluster.ClusterMembershipClient;
+import org.cache.cluster.ClusterTopology;
 import org.cache.cluster.routing.ClusterForwardingClient;
 import org.cache.cluster.routing.RoutedCacheService;
 import org.cache.config.CacheConfig;
@@ -70,6 +74,20 @@ public CacheNode cacheNode(CacheConfig cacheConfig) {
         return cacheConfig.cacheNode();
     }
 
+    @Bean
+    public ClusterMembership clusterMembership(CacheConfig cacheConfig) {
+        if (cacheConfig.clusterInfo() == null) {
+            return new ClusterMembership(null);
+        }
+
+        return new ClusterMembership(new ClusterTopology(
+                0,
+                cacheConfig.clusterInfo().nodes(),
+                cacheConfig.clusterInfo().replicationFactor(),
+                128
+        ));
+    }
+
     @Bean
     public KeyCodec<Object> keyCodec(CacheConfig cacheConfig) {
         return cacheConfig.keyCodec();
@@ -97,56 +115,77 @@ public CacheService<Object> cacheService(Cache<Object> cache, ValueCodecRegistry
     public CacheOperations<Object> routedCacheService(
             CacheService<Object> cacheService,
             CacheNode cacheNode,
-            CacheConfig cacheConfig,
             ClusterForwardingClient forwardingClient,
-            KeyCodec<Object> keyCodec
+            KeyCodec<Object> keyCodec,
+            ClusterMembership clusterMembership
     ) {
-        return new RoutedCacheService<>(cacheService, cacheNode, cacheConfig.clusterInfo(), forwardingClient, keyCodec);
+        return new RoutedCacheService<>(cacheService, cacheNode, forwardingClient, keyCodec, true, clusterMembership);
     }
 
     @Bean
     public CacheOperations<Object> clusterRoutedCacheService(
             CacheService<Object> cacheService,
             CacheNode cacheNode,
-            CacheConfig cacheConfig,
             ClusterForwardingClient forwardingClient,
-            KeyCodec<Object> keyCodec
+            KeyCodec<Object> keyCodec,
+            ClusterMembership clusterMembership
     ) {
         return new RoutedCacheService<>(
                 cacheService,
                 cacheNode,
-                cacheConfig.clusterInfo(),
                 forwardingClient,
                 keyCodec,
-                false
+                false,
+                clusterMembership
         );
     }
 
     @Bean
-    public CommandProcessor<Object> commandProcessor(KeyCodec<Object> keyCodec, CacheOperations<Object> cacheService) {
-        return new CommandProcessor<>(keyCodec, cacheService);
+    public CommandProcessor<Object> commandProcessor(
+            KeyCodec<Object> keyCodec,
+            CacheOperations<Object> cacheService,
+            ClusterMembership clusterMembership,
+            ClusterGossipService clusterGossipService
+    ) {
+        return new CommandProcessor<>(keyCodec, cacheService, clusterMembership, clusterGossipService, false);
     }
 
     @Bean
     public CommandProcessor<Object> clusterCommandProcessor(
             KeyCodec<Object> keyCodec,
-            @Qualifier("clusterRoutedCacheService") CacheOperations<Object> cacheService
+            @Qualifier("clusterRoutedCacheService") CacheOperations<Object> cacheService,
+            ClusterMembership clusterMembership,
+            ClusterGossipService clusterGossipService
     ) {
-        return new CommandProcessor<>(keyCodec, cacheService);
+        return new CommandProcessor<>(keyCodec, cacheService, clusterMembership, clusterGossipService);
     }
 
     @Bean
     public ClusterForwardingClient clusterForwardingClient() {
         return new ClusterForwardingClient();
     }
 
+    @Bean
+    public ClusterMembershipClient clusterMembershipClient(ClusterForwardingClient forwardingClient) {
+        return new ClusterMembershipClient(forwardingClient);
+    }
+
     @Bean
     public ClusterHealthMonitor clusterHealthMonitor(
             CacheNode cacheNode,
-            CacheConfig cacheConfig,
-            ClusterForwardingClient forwardingClient
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient
+    ) {
+        return new ClusterHealthMonitor(cacheNode, clusterMembership, membershipClient);
+    }
+
+    @Bean
+    public ClusterGossipService clusterGossipService(
+            CacheNode cacheNode,
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient
     ) {
-        return new ClusterHealthMonitor(cacheNode, cacheConfig.clusterInfo(), forwardingClient);
+        return new ClusterGossipService(cacheNode, clusterMembership, membershipClient);
     }
 
     @Bean(destroyMethod = "shutdownNow")

src/main/java/org/cache/cluster/CacheInfoException.java

@@ -0,0 +1,7 @@
+package org.cache.cluster;
+
+public class CacheInfoException extends IllegalArgumentException {
+    public CacheInfoException(String message) {
+        super(message);
+    }
+}

src/main/java/org/cache/cluster/ClusterGossipService.java

@@ -0,0 +1,132 @@
+package org.cache.cluster;
+
+import org.springframework.context.SmartLifecycle;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class ClusterGossipService implements SmartLifecycle {
+
+    private static final long INITIAL_DELAY_SECONDS = 3;
+    private static final long GOSSIP_INTERVAL_SECONDS = 3;
+
+    private final CacheNode currentNode;
+    private final ClusterMembership clusterMembership;
+    private final ClusterMembershipClient membershipClient;
+    private final ScheduledExecutorService executor;
+    private volatile boolean running;
+
+    public ClusterGossipService(
+            CacheNode currentNode,
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient
+    ) {
+        this(currentNode, clusterMembership, membershipClient, Executors.newSingleThreadScheduledExecutor());
+    }
+
+    ClusterGossipService(
+            CacheNode currentNode,
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient,
+            ScheduledExecutorService executor
+    ) {
+        this.currentNode = currentNode;
+        this.clusterMembership = clusterMembership;
+        this.membershipClient = membershipClient;
+        this.executor = executor;
+    }
+
+    @Override
+    public void start() {
+        if (running || clusterMembership.currentTopology() == null) {
+            return;
+        }
+
+        running = true;
+        executor.scheduleAtFixedRate(
+                this::gossipOnce,
+                INITIAL_DELAY_SECONDS,
+                GOSSIP_INTERVAL_SECONDS,
+                TimeUnit.SECONDS
+        );
+    }
+
+    @Override
+    public void stop() {
+        running = false;
+        executor.shutdownNow();
+    }
+
+    @Override
+    public boolean isRunning() {
+        return running;
+    }
+
+    public void broadcastTopology() {
+        ClusterTopology topology = clusterMembership.currentTopology();
+        if (topology == null) {
+            return;
+        }
+
+        for (CacheNode peer : topology.nodes()) {
+            if (isNotCurrentNode(peer)) {
+                membershipClient.applyTopology(peer, topology);
+            }
+        }
+    }
+
+    void gossipOnce() {
+        try {
+            ClusterTopology topology = clusterMembership.currentTopology();
+            if (topology == null) {
+                return;
+            }
+
+            for (CacheNode peer : topology.nodes()) {
+                if (isNotCurrentNode(peer)) {
+                    gossipWith(peer);
+                }
+            }
+        } catch (RuntimeException exception) {
+            System.err.println("Cluster gossip failed: " + exception.getMessage());
+        }
+
+    }
+
+    private void gossipWith(CacheNode peer) {
+        ClusterTopology localTopology = clusterMembership.currentTopology();
+        membershipClient.topologyDigest(peer).ifPresent(peerDigest -> {
+            if (peerDigest.version() < localTopology.version()) {
+                membershipClient.applyTopology(peer, localTopology);
+                return;
+            }
+
+            if (peerDigest.version() > localTopology.version()) {
+                applyPeerTopology(peer);
+                return;
+            }
+
+            if (!peerDigest.fingerprint().equals(localTopology.fingerprint())) {
+                resolveSameVersionConflict(peer, localTopology);
+            }
+        });
+    }
+
+    private void resolveSameVersionConflict(CacheNode peer, ClusterTopology localTopology) {
+        boolean appliedPeerTopology = applyPeerTopology(peer);
+        if (!appliedPeerTopology) {
+            membershipClient.applyTopology(peer, localTopology);
+        }
+    }
+
+    private boolean applyPeerTopology(CacheNode peer) {
+        return membershipClient.topology(peer)
+                .map(clusterMembership::applyTopology)
+                .orElse(false);
+    }
+
+    private boolean isNotCurrentNode(CacheNode node) {
+        return !node.getId().equals(currentNode.getId());
+    }
+}

src/main/java/org/cache/cluster/ClusterHealthMonitor.java

@@ -1,6 +1,5 @@
 package org.cache.cluster;
 
-import org.cache.cluster.routing.ClusterForwardingClient;
 import org.springframework.context.SmartLifecycle;
 
 import java.util.HashMap;
@@ -17,35 +16,35 @@ public class ClusterHealthMonitor implements SmartLifecycle {
     private static final long CHECK_INTERVAL_SECONDS = 5;
 
     private final CacheNode currentNode;
-    private final ClusterInfo clusterInfo;
-    private final ClusterForwardingClient forwardingClient;
+    private final ClusterMembership clusterMembership;
+    private final ClusterMembershipClient membershipClient;
     private final Map<String, Integer> failureCounts = new HashMap<>();
     private final ScheduledExecutorService executor;
     private volatile boolean running;
 
     public ClusterHealthMonitor(
             CacheNode currentNode,
-            ClusterInfo clusterInfo,
-            ClusterForwardingClient forwardingClient
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient
     ) {
-        this(currentNode, clusterInfo, forwardingClient, Executors.newSingleThreadScheduledExecutor());
+        this(currentNode, clusterMembership, membershipClient, Executors.newSingleThreadScheduledExecutor());
     }
 
     ClusterHealthMonitor(
             CacheNode currentNode,
-            ClusterInfo clusterInfo,
-            ClusterForwardingClient forwardingClient,
+            ClusterMembership clusterMembership,
+            ClusterMembershipClient membershipClient,
             ScheduledExecutorService executor
     ) {
         this.currentNode = currentNode;
-        this.clusterInfo = clusterInfo;
-        this.forwardingClient = forwardingClient;
+        this.clusterMembership = clusterMembership;
+        this.membershipClient = membershipClient;
         this.executor = executor;
     }
 
     @Override
     public void start() {
-        if (running || clusterInfo == null) {
+        if (running || clusterMembership.currentTopology() == null) {
             return;
         }
 
@@ -70,11 +69,12 @@ public boolean isRunning() {
     }
 
     void checkCluster() {
-        if (clusterInfo == null) {
+        ClusterTopology topology = clusterMembership.currentTopology();
+        if (topology == null) {
             return;
         }
 
-        for (CacheNode node : clusterInfo.nodes()) {
+        for (CacheNode node : topology.nodes()) {
             if (!node.getId().equals(currentNode.getId())) {
                 checkNode(node);
             }
@@ -90,15 +90,15 @@ private void checkClusterSafely() {
     }
 
     private void checkNode(CacheNode node) {
-        if (forwardingClient.ping(node)) {
+        if (membershipClient.ping(node)) {
             failureCounts.remove(node.getId());
-            node.setStatus(NodeStatus.HEALTHY);
+            clusterMembership.markStatus(node.getId(), NodeStatus.HEALTHY);
             return;
         }
 
         int failures = failureCounts.getOrDefault(node.getId(), 0) + 1;
         failureCounts.put(node.getId(), failures);
-        node.setStatus(statusFor(failures));
+        clusterMembership.markStatus(node.getId(), statusFor(failures));
     }
 
     private NodeStatus statusFor(int failures) {