Merge pull request #3044, code review around RouterChain.

Author: beiwei30

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Router.java

@@ -22,7 +22,6 @@
 import org.apache.dubbo.rpc.RpcException;
 
 import java.util.List;
-import java.util.Map;
 
 /**
  * Router. (SPI, Prototype, ThreadSafe)
@@ -43,24 +42,25 @@ public interface Router extends Comparable<Router> {
     /**
      * Filter invokers with current routing rule and only return the invokers that comply with the rule.
      *
-     * @param invokers
+     * @param invokers   invoker list
      * @param url        refer url
-     * @param invocation
+     * @param invocation invocation
      * @return routed invokers
      * @throws RpcException
      */
     <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
 
-    default <T> Map<String, List<Invoker<T>>> preRoute(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
-        return null;
-    }
 
     /**
-     * Each router has a reference of the router chain.
+     * Notify the router the invoker list. Invoker list may change from time to time. This method gives the router a
+     * chance to prepare before {@link Router#route(List, URL, Invocation)} gets called.
      *
-     * @param routerChain
+     * @param invokers invoker list
+     * @param <T>      invoker's type
      */
-    void addRouterChain(RouterChain routerChain);
+    default <T> void notify(List<Invoker<T>> invokers) {
+
+    }
 
     /**
      * To decide whether this router need to execute every time an RPC comes or should only execute when addresses or rule change.
@@ -83,4 +83,4 @@ default <T> Map<String, List<Invoker<T>>> preRoute(List<Invoker<T>> invokers, UR
      * @return
      */
     int getPriority();
-}
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java

@@ -18,7 +18,6 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.common.utils.CollectionUtils;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 
@@ -33,92 +32,78 @@
 public class RouterChain<T> {
 
     // full list of addresses from registry, classified by method name.
-    private List<Invoker<T>> fullInvokers;
+    private List<Invoker<T>> invokers = Collections.emptyList();
     private URL url;
 
     // containing all routers, reconstruct every time 'route://' urls change.
-    private List<Router> routers = new CopyOnWriteArrayList<>();
-    // Fixed router instances: ConfigConditionRouter, TagRouter, e.g., the rule for each instance may change but the instance will never delete or recreate.
-    private List<Router> residentRouters;
+    private volatile List<Router> routers = Collections.emptyList();
 
-    public static <T> RouterChain<T> buildChain(URL url) {
-        RouterChain<T> routerChain = new RouterChain<>(url);
-        List<RouterFactory> extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class).getActivateExtension(url, (String[]) null);
-        List<Router> routers = extensionFactories.stream()
-                .map(factory -> {
-                    Router router = factory.getRouter(url);
-                    router.addRouterChain(routerChain);
-                    return router;
-                }).collect(Collectors.toList());
-        routerChain.setResidentRouters(routers);
-        return routerChain;
-    }
+    // Fixed router instances: ConfigConditionRouter, TagRouter, e.g., the rule for each instance may change but the
+    // instance will never delete or recreate.
+    private List<Router> builtinRouters = Collections.emptyList();
 
-    protected RouterChain(List<Router> routers) {
-        this.routers.addAll(routers);
+    public static <T> RouterChain<T> buildChain(URL url) {
+        return new RouterChain<>(url);
     }
 
-    protected RouterChain(URL url) {
+    private RouterChain(URL url) {
         this.url = url;
+
+        List<RouterFactory> extensionFactories = ExtensionLoader.getExtensionLoader(RouterFactory.class)
+                .getActivateExtension(url, (String[]) null);
+
+        List<Router> routers = extensionFactories.stream()
+                .map(factory -> factory.getRouter(url))
+                .collect(Collectors.toList());
+
+        initWithRouters(routers);
     }
 
     /**
      * the resident routers must being initialized before address notification.
-     *
-     * @param residentRouters
+     * FIXME: this method should not be public
      */
-    public void setResidentRouters(List<Router> residentRouters) {
-        this.residentRouters = residentRouters;
-        this.routers.addAll(residentRouters);
+    public void initWithRouters(List<Router> builtinRouters) {
+        this.builtinRouters = builtinRouters;
+        this.routers = new CopyOnWriteArrayList<>(builtinRouters);
         this.sort();
     }
 
+    public void addRouter(Router router) {
+        this.routers.add(router);
+    }
+
     /**
      * If we use route:// protocol in version before 2.7.0, each URL will generate a Router instance,
      * so we should keep the routers up to date, that is, each time router URLs changes, we should update the routers list,
-     * only keep the residentRouters which are available all the time and the latest notified routers which are generated from URLs.
+     * only keep the builtinRouters which are available all the time and the latest notified routers which are generated from URLs.
      *
-     * @param generatedRouters routers from 'router://' rules in 2.6.x or before.
+     * @param routers routers from 'router://' rules in 2.6.x or before.
      */
-    public void setGeneratedRouters(List<Router> generatedRouters) {
+    public void addRouters(List<Router> routers) {
+        // FIXME will sort cause concurrent problem? since it's kind of a write operation.
         List<Router> newRouters = new CopyOnWriteArrayList<>();
-        newRouters.addAll(residentRouters);
-        newRouters.addAll(generatedRouters);
+        newRouters.addAll(builtinRouters);
+        newRouters.addAll(routers);
         this.routers = newRouters;
-        // FIXME will sort cause concurrent problem? since it's kind of a write operation.
         this.sort();
-       /* if (fullInvokers != null) {
-            this.preRoute(fullInvokers, url, null);
+       /* if (invokers != null) {
+            this.rebuild(invokers, url, null);
         }*/
     }
 
-    public void sort() {
+    private void sort() {
         Collections.sort(routers);
     }
 
-    /**
-     * TODO
-     *
-     * Building of router cache can be triggered from within different threads, for example, registry notification and governance notification.
-     * So this operation should be synchronized.
-     * @param invokers
-     * @param url
-     * @param invocation
-     */
-    public void preRoute(List<Invoker<T>> invokers, URL url, Invocation invocation) {
-        for (Router router : routers) {
-            router.preRoute(invokers, url, invocation);
-        }
-    }
-
     /**
      *
      * @param url
      * @param invocation
      * @return
      */
     public List<Invoker<T>> route(URL url, Invocation invocation) {
-        List<Invoker<T>> finalInvokers = fullInvokers;
+        List<Invoker<T>> finalInvokers = invokers;
         for (Router router : routers) {
 //            if (router.isRuntime()) {
 //                finalInvokers = router.route(finalInvokers, url, invocation);
@@ -128,29 +113,14 @@ public List<Invoker<T>> route(URL url, Invocation invocation) {
         return finalInvokers;
     }
 
-    /**
-     * When any of the router's rule changed, notify the router chain to rebuild cache from scratch.
-     */
-    public void notifyRuleChanged() {
-        if (CollectionUtils.isEmpty(this.fullInvokers)) {
-            return;
-        }
-        preRoute(this.fullInvokers, url, null);
-    }
-
     /**
      * Notify router chain of the initial addresses from registry at the first time.
      * Notify whenever addresses in registry change.
-     *
-     * @param invokers
-     * @param url
      */
-    public void notifyFullInvokers(List<Invoker<T>> invokers, URL url) {
-        setFullMethodInvokers(invokers);
-        preRoute(invokers, url, null);
-    }
-
-    public void setFullMethodInvokers(List<Invoker<T>> fullInvokers) {
-        this.fullInvokers = (fullInvokers == null ? Collections.emptyList() : fullInvokers);
+    public void setInvokers(List<Invoker<T>> invokers) {
+        if (invokers != null) {
+            this.invokers = invokers;
+            routers.forEach(router -> router.notify(invokers));
+        }
     }
 }

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java

@@ -98,7 +98,7 @@ public void setRouterChain(RouterChain<T> routerChain) {
     protected void addRouters(List<Router> routers) {
         // copy list
         routers = routers == null ? new ArrayList<>() : new ArrayList<>(routers);
-        routerChain.setGeneratedRouters(routers);
+        routerChain.addRouters(routers);
     }
 
     public URL getConsumerUrl() {

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java

@@ -87,7 +87,7 @@ public void destroy() {
 
     public void buildRouterChain() {
         RouterChain<T> routerChain = RouterChain.buildChain(getUrl());
-        routerChain.notifyFullInvokers(invokers, getUrl());
+        routerChain.setInvokers(invokers);
         this.setRouterChain(routerChain);
     }
 

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/AbstractRouter.java

@@ -19,10 +19,6 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.configcenter.DynamicConfiguration;
 import org.apache.dubbo.rpc.cluster.Router;
-import org.apache.dubbo.rpc.cluster.RouterChain;
-
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * TODO Extract more code to here if necessary
@@ -31,7 +27,6 @@ public abstract class AbstractRouter implements Router {
     protected int priority;
     protected boolean force = false;
     protected boolean enabled = true;
-    protected List<RouterChain> routerChains = new CopyOnWriteArrayList<>();
     protected URL url;
 
     protected DynamicConfiguration configuration;
@@ -53,12 +48,6 @@ public void setUrl(URL url) {
         this.url = url;
     }
 
-    @Override
-    public void addRouterChain(RouterChain routerChain) {
-        this.routerChains.add(routerChain);
-    }
-
-
     public void setConfiguration(DynamicConfiguration configuration) {
         this.configuration = configuration;
     }

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AbstractConfigConditionRouter.java

@@ -28,7 +28,6 @@
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.cluster.Router;
-import org.apache.dubbo.rpc.cluster.RouterChain;
 import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
 import org.apache.dubbo.rpc.cluster.router.condition.ConditionRouter;
 import org.apache.dubbo.rpc.cluster.router.condition.config.model.BlackWhiteListRule;
@@ -73,8 +72,6 @@ public synchronized void process(ConfigChangeEvent event) {
                         .getValue(), e);
             }
         }
-
-        routerChains.forEach(RouterChain::notifyRuleChanged);
     }
 
     @Override