feat(api): implement routing logic which support dynamically adding/removing broadcast server

Author: BrotherJing

scalable-ot-api/src/main/java/com/brotherjing/service/DiscoveryService.java

@@ -1,8 +1,6 @@
 package com.brotherjing.service;
 
 import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -48,8 +46,8 @@ public void init() {
         });
     }
 
-    public List<ServerEntity> getAllServers() {
-        return new ArrayList<>(cache);
+    Set<ServerEntity> getAllServers() {
+        return cache;
     }
 
     private void addServer(PathChildrenCacheEvent event) {

scalable-ot-api/src/main/java/com/brotherjing/service/RouteService.java

@@ -1,13 +1,17 @@
 package com.brotherjing.service;
 
-import java.util.List;
+import java.util.Set;
+
+import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import com.brotherjing.core.dao.RedisDao;
 import com.brotherjing.core.loadbalance.LoadBalancer;
 import com.brotherjing.core.loadbalance.ServerEntity;
 
+@Slf4j
 @Service
 public class RouteService {
 
@@ -17,8 +21,28 @@ public class RouteService {
     @Autowired
     private LoadBalancer loadBalancer;
 
+    @Autowired
+    private RedisDao redisDao;
+
     public ServerEntity getRoute(String docId) {
-        List<ServerEntity> allServers = discoveryService.getAllServers();
+        Set<ServerEntity> allServers = discoveryService.getAllServers();
+
+        // first check cached route
+        ServerEntity cachedRoute = redisDao.getRoute(docId);
+        if (cachedRoute != null) {
+            if (allServers.contains(cachedRoute)) {
+                // the cached route might differ from the one selected by load balance due to adding new node.
+                // but the old node may still have clients connected to it,
+                // so use the cached route as long as the node is up.
+                return cachedRoute;
+            } else {
+                // this can happen when node is down and the cache has not expired
+                log.warn("Found obsolete route in cache: {}, evicted.", cachedRoute);
+                redisDao.removeRoute(docId);
+            }
+        }
+
+        // if no cache, use load balance to select route
         return loadBalancer.select(allServers, docId);
     }
 }