Condition route v3.1

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

@@ -138,4 +138,12 @@ public interface Constants {
     String RULE_VERSION_V27 = "v2.7";
 
     String RULE_VERSION_V30 = "v3.0";
+
+    String RULE_VERSION_V31 = "v3.1";
+
+    public static final String TRAFFIC_DISABLE_KEY = "trafficDisable";
+    public static final String RATIO_KEY = "ratio";
+    public static final int DefaultRouteRatio = 0;
+    public static final int DefaultRouteConditionSubSetWeight = 100;
+    public static final int DefaultRoutePriority = 0;
 }

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

@@ -29,10 +29,13 @@
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.router.AbstractRouterRule;
 import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
 import org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouter;
+import org.apache.dubbo.rpc.cluster.router.condition.MultiDestConditionRouter;
 import org.apache.dubbo.rpc.cluster.router.condition.config.model.ConditionRouterRule;
 import org.apache.dubbo.rpc.cluster.router.condition.config.model.ConditionRuleParser;
+import org.apache.dubbo.rpc.cluster.router.condition.config.model.MultiDestConditionRouterRule;
 import org.apache.dubbo.rpc.cluster.router.state.AbstractStateRouter;
 import org.apache.dubbo.rpc.cluster.router.state.BitList;
 import org.apache.dubbo.rpc.cluster.router.state.TailStateRouter;
@@ -42,6 +45,7 @@
 import java.util.stream.Collectors;
 
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_RULE_PARSING;
+import static org.apache.dubbo.rpc.cluster.Constants.RULE_VERSION_V31;
 
 /**
  * Abstract router which listens to dynamic configuration
@@ -52,8 +56,11 @@ public abstract class ListenableStateRouter<T> extends AbstractStateRouter<T> im
 
     private static final ErrorTypeAwareLogger logger =
             LoggerFactory.getErrorTypeAwareLogger(ListenableStateRouter.class);
-    private volatile ConditionRouterRule routerRule;
+    private volatile AbstractRouterRule routerRule;
     private volatile List<ConditionStateRouter<T>> conditionRouters = Collections.emptyList();
+
+    //    for v3.1
+    private volatile List<MultiDestConditionRouter<T>> multiDestConditionRouters = Collections.emptyList();
     private final String ruleKey;
 
     public ListenableStateRouter(URL url, String ruleKey) {
@@ -73,6 +80,8 @@ public synchronized void process(ConfigChangedEvent event) {
         if (event.getChangeType().equals(ConfigChangeType.DELETED)) {
             routerRule = null;
             conditionRouters = Collections.emptyList();
+            //          for v3.1
+            multiDestConditionRouters = Collections.emptyList();
         } else {
             try {
                 routerRule = ConditionRuleParser.parse(event.getContent());
@@ -99,7 +108,8 @@ public BitList<Invoker<T>> doRoute(
             Holder<RouterSnapshotNode<T>> nodeHolder,
             Holder<String> messageHolder)
             throws RpcException {
-        if (CollectionUtils.isEmpty(invokers) || conditionRouters.size() == 0) {
+        if (CollectionUtils.isEmpty(invokers)
+                || (conditionRouters.size() == 0 && multiDestConditionRouters.size() == 0)) {
             if (needToPrintMessage) {
                 messageHolder.set(
                         "Directly return. Reason: Invokers from previous router is empty or conditionRouters is empty.");
@@ -112,18 +122,45 @@ public BitList<Invoker<T>> doRoute(
         if (needToPrintMessage) {
             resultMessage = new StringBuilder();
         }
-        for (AbstractStateRouter<T> router : conditionRouters) {
-            invokers = router.route(invokers, url, invocation, needToPrintMessage, nodeHolder);
-            if (needToPrintMessage) {
-                resultMessage.append(messageHolder.get());
+
+        BitList<Invoker<T>> routeResult = invokers;
+        if (routerRule instanceof MultiDestConditionRouterRule
+                || routerRule.getVersion() != null && routerRule.getVersion().startsWith(RULE_VERSION_V31)) {
+            boolean trafficDisable = false;
+            for (MultiDestConditionRouter<T> multiDestConditionRouter : multiDestConditionRouters) {
+                routeResult = multiDestConditionRouter.route(invokers, url, invocation, needToPrintMessage, nodeHolder);
+                if (invokers == routeResult) {
+                    //                    not match or disable to continue next multiDestConditionRouter
+                    continue;
+                } else if (routeResult.size() == 0
+                        && !multiDestConditionRouter.isTrafficDisable()
+                        && !multiDestConditionRouter.isForce()) {
+                    //                    empty but can continue to next multiDestConditionRouter
+                    continue;
+                } else {
+                    trafficDisable = multiDestConditionRouter.isTrafficDisable();
+                    break;
+                }
+            }
+            //            if trafficDisable ignore root.force
+            if (routeResult.size() == 0 && !routerRule.isForce() && !trafficDisable) {
+                routeResult = invokers;
+            }
+        } else {
+            for (AbstractStateRouter<T> router : conditionRouters) {
+                routeResult = router.route(invokers, url, invocation, needToPrintMessage, nodeHolder);
             }
         }
 
+        if (needToPrintMessage) {
+            resultMessage.append(messageHolder.get());
+        }
+
         if (needToPrintMessage) {
             messageHolder.set(resultMessage.toString());
         }
 
-        return invokers;
+        return routeResult;
     }
 
     @Override
@@ -135,15 +172,31 @@ private boolean isRuleRuntime() {
         return routerRule != null && routerRule.isValid() && routerRule.isRuntime();
     }
 
-    private void generateConditions(ConditionRouterRule rule) {
-        if (rule != null && rule.isValid()) {
-            this.conditionRouters = rule.getConditions().stream()
-                    .map(condition ->
-                            new ConditionStateRouter<T>(getUrl(), condition, rule.isForce(), rule.isEnabled()))
-                    .collect(Collectors.toList());
+    private void generateConditions(AbstractRouterRule rule) {
+        if (rule == null || !rule.isValid()) {
+            return;
+        }
+
+        if (rule instanceof ConditionRouterRule) {
+            this.conditionRouters = ((ConditionRouterRule) rule)
+                    .getConditions().stream()
+                            .map(condition ->
+                                    new ConditionStateRouter<T>(getUrl(), condition, rule.isForce(), rule.isEnabled()))
+                            .collect(Collectors.toList());
+
             for (ConditionStateRouter<T> conditionRouter : this.conditionRouters) {
                 conditionRouter.setNextRouter(TailStateRouter.getInstance());
             }
+        } else if (rule instanceof MultiDestConditionRouterRule) {
+            this.multiDestConditionRouters = ((MultiDestConditionRouterRule) rule)
+                    .getConditions().stream()
+                            .map(condition -> new MultiDestConditionRouter<T>(getUrl(), condition, rule.isEnabled()))
+                            .sorted((a, b) -> a.getPriority() - b.getPriority())
+                            .collect(Collectors.toList());
+
+            for (MultiDestConditionRouter<T> conditionRouter : this.multiDestConditionRouters) {
+                conditionRouter.setNextRouter(TailStateRouter.getInstance());
+            }
         }
     }
 

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

@@ -28,7 +28,7 @@ public class ConditionRouterRule extends AbstractRouterRule {
     private List<String> conditions;
 
     @SuppressWarnings("unchecked")
-    public static ConditionRouterRule parseFromMap(Map<String, Object> map) {
+    public static AbstractRouterRule parseFromMap(Map<String, Object> map) {
         ConditionRouterRule conditionRouterRule = new ConditionRouterRule();
         conditionRouterRule.parseFromMap0(map);
 

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

@@ -16,14 +16,21 @@
  */
 package org.apache.dubbo.rpc.cluster.router.condition.config.model;
 
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.rpc.cluster.router.AbstractRouterRule;
 
 import java.util.Map;
 
 import org.yaml.snakeyaml.LoaderOptions;
 import org.yaml.snakeyaml.Yaml;
 import org.yaml.snakeyaml.constructor.SafeConstructor;
 
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_RULE_PARSING;
+import static org.apache.dubbo.rpc.cluster.Constants.CONFIG_VERSION_KEY;
+import static org.apache.dubbo.rpc.cluster.Constants.RULE_VERSION_V31;
+
 /**
  * %YAML1.2
  *
@@ -40,14 +47,35 @@
  */
 public class ConditionRuleParser {
 
-    public static ConditionRouterRule parse(String rawRule) {
+    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ConditionRuleParser.class);
+
+    public static AbstractRouterRule parse(String rawRule) {
+        AbstractRouterRule rule;
         Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
         Map<String, Object> map = yaml.load(rawRule);
-        ConditionRouterRule rule = ConditionRouterRule.parseFromMap(map);
-        rule.setRawRule(rawRule);
-        if (CollectionUtils.isEmpty(rule.getConditions())) {
+        String confVersion = (String) map.get(CONFIG_VERSION_KEY);
+
+        if (confVersion != null && confVersion.toLowerCase().startsWith(RULE_VERSION_V31)) {
+            rule = MultiDestConditionRouterRule.parseFromMap(map);
+            if (CollectionUtils.isEmpty(((MultiDestConditionRouterRule) rule).getConditions())) {
+                rule.setValid(false);
+            }
+        } else if (confVersion != null && confVersion.compareToIgnoreCase(RULE_VERSION_V31) > 0) {
+            logger.warn(
+                    CLUSTER_FAILED_RULE_PARSING,
+                    "Invalid condition config version number.",
+                    "",
+                    "Ignore this configuration. Only " + RULE_VERSION_V31 + " and below are supported in this release");
+            rule = ConditionRouterRule.parseFromMap(map);
             rule.setValid(false);
+        } else {
+            //            for under v3.1
+            rule = ConditionRouterRule.parseFromMap(map);
+            if (CollectionUtils.isEmpty(((ConditionRouterRule) rule).getConditions())) {
+                rule.setValid(false);
+            }
         }
+        rule.setRawRule(rawRule);
 
         return rule;
     }

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

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.router.condition.config.model;
+
+import org.apache.dubbo.rpc.cluster.router.condition.matcher.ConditionMatcher;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.dubbo.rpc.cluster.Constants.DefaultRouteConditionSubSetWeight;
+
+public class ConditionSubSet {
+    private Map<String, ConditionMatcher> condition = new HashMap<>();
+    private Integer subSetWeight;
+
+    public ConditionSubSet() {}
+
+    public ConditionSubSet(Map<String, ConditionMatcher> condition, Integer subSetWeight) {
+        this.condition = condition;
+        this.subSetWeight = subSetWeight;
+        if (subSetWeight <= 0) {
+            this.subSetWeight = DefaultRouteConditionSubSetWeight;
+        }
+    }
+
+    public Map<String, ConditionMatcher> getCondition() {
+        return condition;
+    }
+
+    public void setCondition(Map<String, ConditionMatcher> condition) {
+        this.condition = condition;
+    }
+
+    public Integer getSubSetWeight() {
+        return subSetWeight;
+    }
+
+    public void setSubSetWeight(int subSetWeight) {
+        this.subSetWeight = subSetWeight;
+    }
+
+    @Override
+    public String toString() {
+        return "ConditionSubSet{" + "cond=" + condition + ", subSetWeight=" + subSetWeight + '}';
+    }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.cluster.router.condition.config.model;
+
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
+
+public class Destination<T> {
+    private int weight;
+    private BitList<Invoker<T>> invokers;
+
+    Destination(int weight, BitList<Invoker<T>> invokers) {
+        this.weight = weight;
+        this.invokers = invokers;
+    }
+
+    public int getWeight() {
+        return weight;
+    }
+
+    public void setWeight(int weight) {
+        this.weight = weight;
+    }
+
+    public BitList<Invoker<T>> getInvokers() {
+        return invokers;
+    }
+
+    public void setInvokers(BitList<Invoker<T>> invokers) {
+        this.invokers = invokers;
+    }
+}