You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2021/12/27 08:00:35 UTC
[dubbo] branch 3.0 updated: [3.0] Refactor state router iterator to router chain (#9424)
This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 77e6708 [3.0] Refactor state router iterator to router chain (#9424)
77e6708 is described below
commit 77e6708ebd8d8d2165d004b1a5a74eda60d40c4c
Author: Albumen Kevin <jh...@gmail.com>
AuthorDate: Mon Dec 27 16:00:22 2021 +0800
[3.0] Refactor state router iterator to router chain (#9424)
* [3.0] Refactor state router iterator to router chain
* add asf
* add method
* fix snapshot print
* change to set nextRouter in StateRouter interface
* add doc
* Add logger check
* change RouterChain constructor
* change RouterChain constructor
* Fix Mesh Router
* remove taillist
* fix ut
* override addAll
* Add unit prepare jdk 17 support
* Revert "Add unit prepare jdk 17 support"
This reverts commit 624753832935b0fd2a0fb2f1ebbc9e4367d8fee4.
* refactor message
* add option
---
.../org/apache/dubbo/rpc/cluster/RouterChain.java | 216 ++++++++++++---------
.../rpc/cluster/directory/AbstractDirectory.java | 6 +-
.../rpc/cluster/router/RouterSnapshotNode.java | 121 ++++++++----
.../router/condition/ConditionStateRouter.java | 54 ++++--
.../router/condition/config/AppStateRouter.java | 5 -
.../condition/config/AppStateRouterFactory.java | 3 +-
.../condition/config/ListenableStateRouter.java | 36 ++--
.../router/condition/config/ServiceRouter.java | 5 -
.../config/ServiceStateRouterFactory.java | 3 +-
.../cluster/router/mesh/route/MeshRuleRouter.java | 22 ++-
.../router/mesh/route/StandardMeshRuleRouter.java | 10 -
.../cluster/router/mock/MockInvokersSelector.java | 40 ++--
.../router/mock/MockStateRouterFactory.java | 2 +-
.../cluster/router/script/ScriptStateRouter.java | 19 +-
.../cluster/router/state/AbstractStateRouter.java | 130 ++++++++++++-
.../dubbo/rpc/cluster/router/state/BitList.java | 32 ++-
.../rpc/cluster/router/state/StateRouter.java | 41 ++--
.../cluster/router/state/StateRouterResult.java | 83 --------
...stractStateRouter.java => TailStateRouter.java} | 56 +++---
.../rpc/cluster/router/tag/TagStateRouter.java | 51 +++--
...bbo.rpc.cluster.router.state.StateRouterFactory | 1 -
.../rpc/cluster/directory/StaticDirectoryTest.java | 3 +-
.../router/condition/ConditionRouterTest.java | 45 ++---
.../cluster/router/file/FileRouterEngineTest.java | 3 +-
.../router/mesh/route/MeshRuleRouterTest.java | 59 +++---
.../router/mock/MockInvokersSelectorTest.java | 5 +-
.../router/script/ScriptStateRouterTest.java | 9 +-
.../rpc/cluster/router/state/BitListTest.java | 20 ++
...bbo.rpc.cluster.router.state.StateRouterFactory | 1 +
.../router/state/StateRouterFactory$Adaptive.java | 3 +-
.../org/apache/dubbo/rpc/RpcServiceContext.java | 10 +
31 files changed, 642 insertions(+), 452 deletions(-)
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
index a679fc7..ace0765 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
@@ -22,15 +22,17 @@ import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.cluster.router.RouterResult;
import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
-import org.apache.dubbo.rpc.cluster.router.state.StateRouterResult;
+import org.apache.dubbo.rpc.cluster.router.state.TailStateRouter;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.util.ArrayList;
@@ -63,8 +65,9 @@ public class RouterChain<T> {
*/
private volatile List<Router> builtinRouters = Collections.emptyList();
- private volatile List<StateRouter<T>> builtinStateRouters = Collections.emptyList();
- private volatile List<StateRouter<T>> stateRouters = Collections.emptyList();
+ private volatile StateRouter<T> headStateRouter;
+
+ private volatile List<StateRouter<T>> stateRouters;
/**
* Should continue route if current router's result is empty
@@ -72,10 +75,6 @@ public class RouterChain<T> {
private final boolean shouldFailFast;
public static <T> RouterChain<T> buildChain(Class<T> interfaceClass, URL url) {
- return new RouterChain<>(interfaceClass, url);
- }
-
- private RouterChain(Class<T> interfaceClass, URL url) {
ModuleModel moduleModel = url.getOrDefaultModuleModel();
List<RouterFactory> extensionFactories = moduleModel.getExtensionLoader(RouterFactory.class)
@@ -86,21 +85,36 @@ public class RouterChain<T> {
.sorted(Router::compareTo)
.collect(Collectors.toList());
- initWithRouters(routers);
-
- List<StateRouterFactory> extensionStateRouterFactories = moduleModel
+ List<StateRouter<T>> stateRouters = moduleModel
.getExtensionLoader(StateRouterFactory.class)
- .getActivateExtension(url, ROUTER_KEY);
-
- List<StateRouter<T>> stateRouters = extensionStateRouterFactories.stream()
+ .getActivateExtension(url, ROUTER_KEY)
+ .stream()
.map(factory -> factory.getRouter(interfaceClass, url))
- .sorted(StateRouter::compareTo)
.collect(Collectors.toList());
- // init state routers
+
+ boolean shouldFailFast = Boolean.parseBoolean(ConfigurationUtils.getProperty(moduleModel, Constants.SHOULD_FAIL_FAST_KEY, "true"));
+
+ return new RouterChain<>(routers, stateRouters, shouldFailFast);
+ }
+
+ public RouterChain(List<Router> routers, List<StateRouter<T>> stateRouters, boolean shouldFailFast) {
+ initWithRouters(routers);
+
initWithStateRouters(stateRouters);
- this.shouldFailFast = Boolean.parseBoolean(ConfigurationUtils.getProperty(moduleModel, Constants.SHOULD_FAIL_FAST_KEY, "true"));
+ this.shouldFailFast = shouldFailFast;
+ }
+
+ private void initWithStateRouters(List<StateRouter<T>> stateRouters) {
+ StateRouter<T> stateRouter = TailStateRouter.getInstance();
+ for (int i = stateRouters.size() - 1; i >= 0; i--) {
+ StateRouter<T> nextStateRouter = stateRouters.get(i);
+ nextStateRouter.setNextRouter(stateRouter);
+ stateRouter = nextStateRouter;
+ }
+ this.headStateRouter = stateRouter;
+ this.stateRouters = Collections.unmodifiableList(stateRouters);
}
/**
@@ -113,19 +127,6 @@ public class RouterChain<T> {
}
/**
- * the resident routers must being initialized before address notification.
- * only for ut
- */
- public void initWithStateRouters(List<StateRouter<T>> builtinRouters) {
- this.builtinStateRouters = builtinRouters;
- setStateRouters(builtinStateRouters);
- }
-
- private void setStateRouters(List<StateRouter<T>> stateRouters) {
- this.stateRouters = new LinkedList<>(stateRouters);
- }
-
- /**
* 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 builtinRouters which are available all the time and the latest notified routers which are generated
@@ -141,47 +142,38 @@ public class RouterChain<T> {
this.routers = newRouters;
}
- public void addStateRouters(List<StateRouter<T>> stateRouters) {
- List<StateRouter<T>> newStateRouters = new LinkedList<>();
- newStateRouters.addAll(builtinStateRouters);
- newStateRouters.addAll(stateRouters);
- CollectionUtils.sort(newStateRouters);
- setStateRouters(newStateRouters);
- }
-
public List<Router> getRouters() {
return routers;
}
- public List<StateRouter<T>> getStateRouters() {
- return stateRouters;
+ public StateRouter<T> getHeadStateRouter() {
+ return headStateRouter;
}
- /**
- * @param url
- * @param invocation
- * @return
- */
public List<Invoker<T>> route(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
+ if (RpcContext.getServiceContext().isNeedPrintRouterSnapshot()) {
+ return routeAndPrint(url, availableInvokers, invocation);
+ } else {
+ return simpleRoute(url, availableInvokers, invocation);
+ }
+ }
+ public List<Invoker<T>> routeAndPrint(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
+ RouterSnapshotNode<T> snapshot = buildRouterSnapshot(url, availableInvokers, invocation);
+ logRouterSnapshot(url, invocation, snapshot);
+ return snapshot.getChainOutputInvokers();
+ }
+
+ public List<Invoker<T>> simpleRoute(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
BitList<Invoker<T>> resultInvokers = availableInvokers.clone();
// 1. route state router
- for (StateRouter<T> stateRouter : stateRouters) {
- StateRouterResult<Invoker<T>> routeResult = stateRouter.route(resultInvokers, url, invocation, false);
- resultInvokers = routeResult.getResult();
- if (resultInvokers.isEmpty() && shouldFailFast) {
- printRouterSnapshot(url, availableInvokers, invocation);
- return BitList.emptyList();
- }
-
- // stop continue routing
- if (!routeResult.isNeedContinueRoute()) {
- return routeResult.getResult();
- }
+ resultInvokers = headStateRouter.route(resultInvokers, url, invocation, false, null);
+ if (resultInvokers.isEmpty() && (shouldFailFast || routers.isEmpty())) {
+ printRouterSnapshot(url, availableInvokers, invocation);
+ return BitList.emptyList();
}
-
if (routers.isEmpty()) {
return resultInvokers;
}
@@ -214,7 +206,9 @@ public class RouterChain<T> {
* store each router's input and output, log out if empty
*/
private void printRouterSnapshot(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
- logRouterSnapshot(url, invocation, buildRouterSnapshot(url, availableInvokers, invocation));
+ if (logger.isWarnEnabled()) {
+ logRouterSnapshot(url, invocation, buildRouterSnapshot(url, availableInvokers, invocation));
+ }
}
/**
@@ -222,68 +216,85 @@ public class RouterChain<T> {
*/
public RouterSnapshotNode<T> buildRouterSnapshot(URL url, BitList<Invoker<T>> availableInvokers, Invocation invocation) {
BitList<Invoker<T>> resultInvokers = availableInvokers.clone();
- RouterSnapshotNode<T> snapshotNode = new RouterSnapshotNode<T>("Parent", resultInvokers.size());
- snapshotNode.setOutputInvokers(resultInvokers.clone());
+ RouterSnapshotNode<T> parentNode = new RouterSnapshotNode<T>("Parent", resultInvokers.clone());
+ parentNode.setNodeOutputInvokers(resultInvokers.clone());
// 1. route state router
- for (StateRouter stateRouter : stateRouters) {
- BitList<Invoker<T>> inputInvokers = resultInvokers.clone();
+ Holder<RouterSnapshotNode<T>> nodeHolder = new Holder<>();
+ nodeHolder.set(parentNode);
- RouterSnapshotNode<T> currentNode = new RouterSnapshotNode<T>(stateRouter.getClass().getSimpleName(), inputInvokers.size());
- snapshotNode.appendNode(currentNode);
+ resultInvokers = headStateRouter.route(resultInvokers, url, invocation, true, nodeHolder);
- StateRouterResult<Invoker<T>> routeResult = stateRouter.route(inputInvokers, url, invocation, true);
- resultInvokers = routeResult.getResult();
- String routerMessage = routeResult.getMessage();
-
- currentNode.setOutputInvokers(resultInvokers);
- currentNode.setRouterMessage(routerMessage);
-
- // result is empty, log out
- if (resultInvokers.isEmpty() && shouldFailFast) {
- return snapshotNode;
- }
-
- if (!routeResult.isNeedContinueRoute()) {
- return snapshotNode;
- }
+ // result is empty, log out
+ if (routers.isEmpty() || (resultInvokers.isEmpty() && shouldFailFast)) {
+ parentNode.setChainOutputInvokers(resultInvokers.clone());
+ return parentNode;
}
+ RouterSnapshotNode<T> commonRouterNode = new RouterSnapshotNode<T>("CommonRouter", resultInvokers.clone());
+ parentNode.appendNode(commonRouterNode);
List<Invoker<T>> commonRouterResult = resultInvokers;
// 2. route common router
for (Router router : routers) {
// Copy resultInvokers to a arrayList. BitList not support
List<Invoker<T>> inputInvokers = new ArrayList<>(commonRouterResult);
- RouterSnapshotNode<T> currentNode = new RouterSnapshotNode<T>(router.getClass().getSimpleName(), inputInvokers.size());
- snapshotNode.appendNode(currentNode);
+ RouterSnapshotNode<T> currentNode = new RouterSnapshotNode<T>(router.getClass().getSimpleName(), inputInvokers);
+
+ // append to router node chain
+ commonRouterNode.appendNode(currentNode);
+ commonRouterNode = currentNode;
RouterResult<Invoker<T>> routeStateResult = router.route(inputInvokers, url, invocation, true);
List<Invoker<T>> routeResult = routeStateResult.getResult();
String routerMessage = routeStateResult.getMessage();
- currentNode.setOutputInvokers(routeResult);
+ currentNode.setNodeOutputInvokers(routeResult);
currentNode.setRouterMessage(routerMessage);
+ commonRouterResult = routeResult;
+
// result is empty, log out
if (CollectionUtils.isEmpty(routeResult) && shouldFailFast) {
- return snapshotNode;
- } else {
- commonRouterResult = routeResult;
+ break;
}
if (!routeStateResult.isNeedContinueRoute()) {
- return snapshotNode;
+ break;
+ }
+ }
+ commonRouterNode.setChainOutputInvokers(commonRouterNode.getNodeOutputInvokers());
+
+ // 3. set router chain output reverse
+ RouterSnapshotNode<T> currentNode = commonRouterNode;
+ while (currentNode != null){
+ RouterSnapshotNode<T> parent = currentNode.getParentNode();
+ if (parent != null) {
+ // common router only has one child invoke
+ parent.setChainOutputInvokers(currentNode.getChainOutputInvokers());
}
+ currentNode = parent;
}
- return snapshotNode;
+ return parentNode;
}
private void logRouterSnapshot(URL url, Invocation invocation, RouterSnapshotNode<T> snapshotNode) {
- logger.warn("No provider available after route for the service " + url.getServiceKey()
- + " from registry " + url.getAddress()
- + " on the consumer " + NetUtils.getLocalHost()
- + " using the dubbo version " + Version.getVersion() + ". Router snapshot is below: \n" + snapshotNode.toString());
+ if (snapshotNode.getChainOutputInvokers() == null ||
+ snapshotNode.getChainOutputInvokers().isEmpty()) {
+ if (logger.isWarnEnabled()) {
+ logger.warn("No provider available after route for the service " + url.getServiceKey()
+ + " from registry " + url.getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion() + ". Router snapshot is below: \n" + snapshotNode.toString());
+ }
+ } else {
+ if (logger.isInfoEnabled()) {
+ logger.info("Router snapshot service " + url.getServiceKey()
+ + " from registry " + url.getAddress()
+ + " on the consumer " + NetUtils.getLocalHost()
+ + " using the dubbo version " + Version.getVersion() + " is below: \n" + snapshotNode.toString());
+ }
+ }
}
/**
@@ -296,6 +307,22 @@ public class RouterChain<T> {
stateRouters.forEach(router -> router.notify(this.invokers));
}
+ /**
+ * for uts only
+ */
+ @Deprecated
+ public void setHeadStateRouter(StateRouter<T> headStateRouter) {
+ this.headStateRouter = headStateRouter;
+ }
+
+ /**
+ * for uts only
+ */
+ @Deprecated
+ public List<StateRouter<T>> getStateRouters() {
+ return stateRouters;
+ }
+
public void destroy() {
invokers = BitList.emptyList();
for (Router router : routers) {
@@ -308,15 +335,14 @@ public class RouterChain<T> {
routers = Collections.emptyList();
builtinRouters = Collections.emptyList();
- for (StateRouter router : stateRouters) {
+ for (StateRouter<T> router : stateRouters) {
try {
router.stop();
} catch (Exception e) {
logger.error("Error trying to stop stateRouter " + router.getClass(), e);
}
}
- setStateRouters(Collections.emptyList());
- builtinStateRouters = Collections.emptyList();
+ stateRouters = Collections.emptyList();
+ headStateRouter = TailStateRouter.getInstance();
}
-
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
index cbce9e4..af9ecb0 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
@@ -188,9 +188,9 @@ public abstract class AbstractDirectory<T> implements Directory<T> {
List<Invoker<T>> routedResult = doList(availableInvokers, invocation);
if (routedResult.isEmpty()) {
logger.warn("No provider available after connectivity filter for the service " + getConsumerUrl().getServiceKey()
- + " all validInvokers' size: " + validInvokers.size()
- + "/ all routed invokers' size: " + routedResult.size()
- + "/ all invokers' size: " + invokers.size()
+ + " All validInvokers' size: " + validInvokers.size()
+ + " All routed invokers' size: " + routedResult.size()
+ + " All invokers' size: " + invokers.size()
+ " from registry " + getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost()
+ " using the dubbo version " + Version.getVersion() + ".");
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java
index 28ac112..d8f21a9 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/RouterSnapshotNode.java
@@ -17,22 +17,39 @@
package org.apache.dubbo.rpc.cluster.router;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
+import java.util.ArrayList;
+import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
public class RouterSnapshotNode<T> {
private final String name;
private final int beforeSize;
- private int afterSize;
+ private int nodeOutputSize;
+ private int chainOutputSize;
private String routerMessage;
- private List<Invoker<T>> outputInvokers;
- private RouterSnapshotNode<T> nextNode;
+ private final List<Invoker<T>> inputInvokers;
+ private List<Invoker<T>> nodeOutputInvokers;
+ private List<Invoker<T>> chainOutputInvokers;
+ private final List<RouterSnapshotNode<T>> nextNode = new LinkedList<>();
+ private RouterSnapshotNode<T> parentNode;
- public RouterSnapshotNode(String name, int beforeSize) {
+ public RouterSnapshotNode(String name, List<Invoker<T>> inputInvokers) {
this.name = name;
- this.beforeSize = beforeSize;
+ this.beforeSize = inputInvokers.size();
+ if (inputInvokers instanceof BitList) {
+ this.inputInvokers = inputInvokers;
+ } else {
+ this.inputInvokers = new ArrayList<>(5);
+ for (int i = 0; i < Math.min(5, beforeSize); i++) {
+ this.inputInvokers.add(inputInvokers.get(i));
+ }
+ }
+ this.nodeOutputSize = 0;
}
public String getName() {
@@ -43,8 +60,8 @@ public class RouterSnapshotNode<T> {
return beforeSize;
}
- public int getAfterSize() {
- return afterSize;
+ public int getNodeOutputSize() {
+ return nodeOutputSize;
}
public String getRouterMessage() {
@@ -55,29 +72,39 @@ public class RouterSnapshotNode<T> {
this.routerMessage = routerMessage;
}
- public List<Invoker<T>> getOutputInvokers() {
- return outputInvokers;
+ public List<Invoker<T>> getNodeOutputInvokers() {
+ return nodeOutputInvokers;
+ }
+
+ public void setNodeOutputInvokers(List<Invoker<T>> outputInvokers) {
+ this.nodeOutputInvokers = outputInvokers;
+ this.nodeOutputSize = outputInvokers == null ? 0 : outputInvokers.size();
}
- public void setOutputInvokers(List<Invoker<T>> outputInvokers) {
- this.outputInvokers = outputInvokers;
- this.afterSize = outputInvokers == null ? 0 : outputInvokers.size();
+ public void setChainOutputInvokers(List<Invoker<T>> outputInvokers) {
+ this.chainOutputInvokers = outputInvokers;
+ this.chainOutputSize = outputInvokers == null ? 0 : outputInvokers.size();
}
- public RouterSnapshotNode<T> getNextNode() {
+ public int getChainOutputSize() {
+ return chainOutputSize;
+ }
+
+ public List<Invoker<T>> getChainOutputInvokers() {
+ return chainOutputInvokers;
+ }
+
+ public List<RouterSnapshotNode<T>> getNextNode() {
return nextNode;
}
+ public RouterSnapshotNode<T> getParentNode() {
+ return parentNode;
+ }
+
public void appendNode(RouterSnapshotNode<T> nextNode) {
- if (this.nextNode != null) {
- RouterSnapshotNode<T> node = this.nextNode;
- while (node.nextNode != null) {
- node = node.nextNode;
- }
- node.nextNode = nextNode;
- } else {
- this.nextNode = nextNode;
- }
+ this.nextNode.add(nextNode);
+ nextNode.parentNode = this;
}
@Override
@@ -90,27 +117,49 @@ public class RouterSnapshotNode<T> {
stringBuilder.append("[ ")
.append(name)
.append(" ")
- .append("invokers: ")
- .append(beforeSize).append(" -> ").append(afterSize)
- .append(" ")
+ .append("(Input: ").append(beforeSize).append(") ")
+ .append("(Current Node Output: ").append(nodeOutputSize).append(") ")
+ .append("(Chain Node Output: ").append(chainOutputSize).append(")")
+ .append(routerMessage == null ? "" : " Router message: ")
.append(routerMessage == null ? "" : routerMessage)
- .append("] ")
- .append(outputInvokers == null ? "" :
- outputInvokers.subList(0, Math.min(5, outputInvokers.size()))
- .stream()
- .map(Invoker::getUrl)
- .map(URL::getAddress)
- .collect(Collectors.joining(",")));
-
- if (outputInvokers != null && outputInvokers.size() > 5) {
+ .append(" ] ");
+ if (level == 1) {
+ stringBuilder.append("Input: ")
+ .append(CollectionUtils.isEmpty(inputInvokers) ? "Empty" :
+ inputInvokers.subList(0, Math.min(5, inputInvokers.size()))
+ .stream()
+ .map(Invoker::getUrl)
+ .map(URL::getAddress)
+ .collect(Collectors.joining(",")))
+ .append(" -> ");
+
+ stringBuilder.append("Chain Node Output: ")
+ .append(CollectionUtils.isEmpty(chainOutputInvokers) ? "Empty" :
+ chainOutputInvokers.subList(0, Math.min(5, chainOutputInvokers.size()))
+ .stream()
+ .map(Invoker::getUrl)
+ .map(URL::getAddress)
+ .collect(Collectors.joining(",")));
+ } else {
+ stringBuilder.append("Current Node Output: ")
+ .append(CollectionUtils.isEmpty(nodeOutputInvokers) ? "Empty" :
+ nodeOutputInvokers.subList(0, Math.min(5, nodeOutputInvokers.size()))
+ .stream()
+ .map(Invoker::getUrl)
+ .map(URL::getAddress)
+ .collect(Collectors.joining(",")));
+ }
+
+
+ if (nodeOutputInvokers != null && nodeOutputInvokers.size() > 5) {
stringBuilder.append("...");
}
- if (nextNode != null) {
+ for (RouterSnapshotNode<T> node : nextNode) {
stringBuilder.append("\n");
for (int i = 0; i < level; i++) {
stringBuilder.append(" ");
}
- stringBuilder.append(nextNode.toString(level + 1));
+ stringBuilder.append(node.toString(level + 1));
}
return stringBuilder.toString();
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionStateRouter.java
index a96a579..92cee81 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionStateRouter.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
@@ -27,10 +28,10 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Constants;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.condition.config.AppStateRouter;
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.StateRouterResult;
import java.text.ParseException;
import java.util.HashMap;
@@ -46,7 +47,6 @@ import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METHOD_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.ADDRESS_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.FORCE_KEY;
-import static org.apache.dubbo.rpc.cluster.Constants.PRIORITY_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.RULE_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.RUNTIME_KEY;
@@ -81,7 +81,6 @@ public class ConditionStateRouter<T> extends AbstractStateRouter<T> {
public ConditionStateRouter(URL url) {
super(url);
this.setUrl(url);
- this.setPriority(url.getParameter(PRIORITY_KEY, 0));
this.setForce(url.getParameter(FORCE_KEY, false));
this.enabled = url.getParameter(ENABLED_KEY, true);
if (enabled) {
@@ -179,44 +178,59 @@ public class ConditionStateRouter<T> extends AbstractStateRouter<T> {
}
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url,
- Invocation invocation, boolean needToPrintMessage) throws RpcException {
-
+ protected BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder,
+ Holder<String> messageHolder) throws RpcException {
if (!enabled) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly return. Reason: ConditionRouter disabled." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly return. Reason: ConditionRouter disabled.");
+ }
+ return invokers;
}
if (CollectionUtils.isEmpty(invokers)) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly return. Reason: Invokers from previous router is empty." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly return. Reason: Invokers from previous router is empty.");
+ }
+ return invokers;
}
try {
if (!matchWhen(url, invocation)) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly return. Reason: WhenCondition not match." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly return. Reason: WhenCondition not match.");
+ }
+ return invokers;
}
if (thenCondition == null) {
logger.warn("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey());
- return new StateRouterResult<>(BitList.emptyList(),
- needToPrintMessage ? "Empty return. Reason: ThenCondition is empty." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Empty return. Reason: ThenCondition is empty.");
+ }
+ return BitList.emptyList();
}
BitList<Invoker<T>> result = invokers.clone();
result.removeIf(invoker -> !matchThen(invoker.getUrl(), url));
if (!result.isEmpty()) {
- return new StateRouterResult<>(result,
- needToPrintMessage ? "Match return." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Match return.");
+ }
+ return result;
} else if (this.isForce()) {
logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey() + ", router: " + url.getParameterAndDecoded(RULE_KEY));
- return new StateRouterResult<>(result,
- needToPrintMessage ? "Empty return. Reason: Empty result from condition and condition is force." : null);
+
+ if (needToPrintMessage) {
+ messageHolder.set("Empty return. Reason: Empty result from condition and condition is force.");
+ }
+ return result;
}
} catch (Throwable t) {
logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t);
}
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly return. Reason: Error occurred ( or result is empty )." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly return. Reason: Error occurred ( or result is empty ).");
+ }
+ return invokers;
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouter.java
index a8117d7..3e483f3 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouter.java
@@ -23,13 +23,8 @@ import org.apache.dubbo.common.URL;
*/
public class AppStateRouter<T> extends ListenableStateRouter<T> {
public static final String NAME = "APP_ROUTER";
- /**
- * AppRouter should after ServiceRouter
- */
- private static final int APP_ROUTER_DEFAULT_PRIORITY = 150;
public AppStateRouter(URL url) {
super(url, url.getApplication());
- this.setPriority(APP_ROUTER_DEFAULT_PRIORITY);
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouterFactory.java
index e35340a..ecfb222 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouterFactory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/AppStateRouterFactory.java
@@ -23,8 +23,9 @@ import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
/**
* Application level router factory
+ * AppRouter should after ServiceRouter
*/
-@Activate(order = 200)
+@Activate(order = 150)
public class AppStateRouterFactory implements StateRouterFactory {
public static final String NAME = "app";
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableStateRouter.java
index 9ca3eb3..2a2b309 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ListenableStateRouter.java
@@ -24,17 +24,18 @@ import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.StringUtils;
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.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouter;
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.state.AbstractStateRouter;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
-import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
-import org.apache.dubbo.rpc.cluster.router.state.StateRouterResult;
+import org.apache.dubbo.rpc.cluster.router.state.TailStateRouter;
import java.util.Collections;
import java.util.List;
@@ -49,7 +50,7 @@ public abstract class ListenableStateRouter<T> extends AbstractStateRouter<T> im
private static final Logger logger = LoggerFactory.getLogger(ListenableStateRouter.class);
private volatile ConditionRouterRule routerRule;
- private volatile List<ConditionStateRouter> conditionRouters = Collections.emptyList();
+ private volatile List<ConditionStateRouter<T>> conditionRouters = Collections.emptyList();
private String ruleKey;
public ListenableStateRouter(URL url, String ruleKey) {
@@ -81,11 +82,13 @@ public abstract class ListenableStateRouter<T> extends AbstractStateRouter<T> im
}
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url,
- Invocation invocation, boolean needToPrintMessage) throws RpcException {
+ public BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder, Holder<String> messageHolder) throws RpcException {
if (CollectionUtils.isEmpty(invokers) || conditionRouters.size() == 0) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly return. Reason: Invokers from previous router is empty or conditionRouters is empty." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly return. Reason: Invokers from previous router is empty or conditionRouters is empty.");
+ }
+ return invokers;
}
// We will check enabled status inside each router.
@@ -93,16 +96,18 @@ public abstract class ListenableStateRouter<T> extends AbstractStateRouter<T> im
if (needToPrintMessage) {
resultMessage = new StringBuilder();
}
- for (StateRouter router : conditionRouters) {
- StateRouterResult<Invoker<T>> routerResult = router.route(invokers, url, invocation, needToPrintMessage);
- invokers = routerResult.getResult();
+ for (AbstractStateRouter<T> router : conditionRouters) {
+ invokers = router.route(invokers, url, invocation, needToPrintMessage, nodeHolder);
if (needToPrintMessage) {
- resultMessage.append(routerResult.getMessage());
+ resultMessage.append(messageHolder.get());
}
}
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? resultMessage.toString() : null);
+ if (needToPrintMessage) {
+ messageHolder.set(resultMessage.toString());
+ }
+
+ return invokers;
}
@Override
@@ -118,8 +123,11 @@ public abstract class ListenableStateRouter<T> extends AbstractStateRouter<T> im
if (rule != null && rule.isValid()) {
this.conditionRouters = rule.getConditions()
.stream()
- .map(condition -> new ConditionStateRouter(getUrl(), condition, rule.isForce(), rule.isEnabled()))
+ .map(condition -> new ConditionStateRouter<T>(getUrl(), condition, rule.isForce(), rule.isEnabled()))
.collect(Collectors.toList());
+ for (ConditionStateRouter<T> conditionRouter : this.conditionRouters) {
+ conditionRouter.setNextRouter(TailStateRouter.getInstance());
+ }
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceRouter.java
index 1f24a87..82bea8c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceRouter.java
@@ -24,13 +24,8 @@ import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
*/
public class ServiceRouter<T> extends ListenableStateRouter<T> {
public static final String NAME = "SERVICE_ROUTER";
- /**
- * ServiceRouter should before AppRouter
- */
- private static final int SERVICE_ROUTER_DEFAULT_PRIORITY = 140;
public ServiceRouter(URL url) {
super(url, DynamicConfiguration.getRuleKey(url));
- this.setPriority(SERVICE_ROUTER_DEFAULT_PRIORITY);
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceStateRouterFactory.java
index a969fd3..965e3bb 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceStateRouterFactory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/condition/config/ServiceStateRouterFactory.java
@@ -23,8 +23,9 @@ import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
/**
* Service level router factory
+ * ServiceRouter should before AppRouter
*/
-@Activate(order = 300)
+@Activate(order = 140)
public class ServiceStateRouterFactory extends CacheableStateRouterFactory {
public static final String NAME = "service";
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
index a2c337a..3ef8c4d 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouter.java
@@ -20,11 +20,13 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.common.utils.StringUtils;
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.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.VsDestinationGroup;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.destination.DestinationRule;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.DubboMatchRequest;
@@ -39,7 +41,6 @@ import org.apache.dubbo.rpc.cluster.router.mesh.util.MeshRuleListener;
import org.apache.dubbo.rpc.cluster.router.mesh.util.TracingContextProvider;
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.StateRouterResult;
import java.util.Collections;
import java.util.HashMap;
@@ -76,10 +77,15 @@ public abstract class MeshRuleRouter<T> extends AbstractStateRouter<T> implement
}
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage) throws RpcException {
+ protected BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder,
+ Holder<String> messageHolder) throws RpcException {
MeshRuleCache<T> ruleCache = this.meshRuleCache;
if (!ruleCache.containsRule()) {
- return new StateRouterResult<>(invokers, needToPrintMessage ? "MeshRuleCache has not been built. Skip route." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("MeshRuleCache has not been built. Skip route.");
+ }
+ return invokers;
}
BitList<Invoker<T>> result = new BitList<>(invokers.getOriginList(), true, invokers.getTailList());
@@ -106,10 +112,16 @@ public abstract class MeshRuleRouter<T> extends AbstractStateRouter<T> implement
// empty protection
if (result.isEmpty()) {
- return new StateRouterResult<>(invokers, needToPrintMessage ? "Empty protection after routed." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Empty protection after routed.");
+ }
+ return invokers;
}
- return new StateRouterResult<>(invokers.and(result), needToPrintMessage ? stringBuilder.toString() : null);
+ if (needToPrintMessage) {
+ messageHolder.set(stringBuilder.toString());
+ }
+ return invokers.and(result);
}
/**
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/StandardMeshRuleRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/StandardMeshRuleRouter.java
index 8da1547..5f891a6 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/StandardMeshRuleRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/StandardMeshRuleRouter.java
@@ -30,14 +30,4 @@ public class StandardMeshRuleRouter<T> extends MeshRuleRouter<T> {
public String ruleSuffix() {
return STANDARD_ROUTER_KEY;
}
-
- @Override
- public boolean isForce() {
- return false;
- }
-
- @Override
- public int getPriority() {
- return -500;
- }
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
index b70a4c4..4654fd3 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelector.java
@@ -18,12 +18,13 @@ package org.apache.dubbo.rpc.cluster.router.mock;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
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.RouterSnapshotNode;
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.StateRouterResult;
import static org.apache.dubbo.rpc.cluster.Constants.INVOCATION_NEED_MOCK;
import static org.apache.dubbo.rpc.cluster.Constants.MOCK_PROTOCOL;
@@ -35,39 +36,48 @@ import static org.apache.dubbo.rpc.cluster.Constants.MOCK_PROTOCOL;
public class MockInvokersSelector<T> extends AbstractStateRouter<T> {
public static final String NAME = "MOCK_ROUTER";
- private static final int MOCK_INVOKERS_DEFAULT_PRIORITY = -100;
private volatile BitList<Invoker<T>> normalInvokers = BitList.emptyList();
private volatile BitList<Invoker<T>> mockedInvokers = BitList.emptyList();
public MockInvokersSelector(URL url) {
super(url);
- this.setPriority(MOCK_INVOKERS_DEFAULT_PRIORITY);
}
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url,
- Invocation invocation, boolean needToPrintMessage) throws RpcException {
+ protected BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder,
+ Holder<String> messageHolder) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Empty invokers. Directly return." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Empty invokers. Directly return.");
+ }
+ return invokers;
}
if (invocation.getObjectAttachments() == null) {
- return new StateRouterResult<>(invokers.and(normalInvokers),
- needToPrintMessage ? "ObjectAttachments from invocation are null. Return normal Invokers." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("ObjectAttachments from invocation are null. Return normal Invokers.");
+ }
+ return invokers.and(normalInvokers);
} else {
String value = (String) invocation.getObjectAttachments().get(INVOCATION_NEED_MOCK);
if (value == null) {
- return new StateRouterResult<>(invokers.and(normalInvokers),
- needToPrintMessage ? "invocation.need.mock not set. Return normal Invokers." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("invocation.need.mock not set. Return normal Invokers.");
+ }
+ return invokers.and(normalInvokers);
} else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
- return new StateRouterResult<>(invokers.and(mockedInvokers),
- needToPrintMessage ? "invocation.need.mock is true. Return mocked Invokers." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("invocation.need.mock is true. Return mocked Invokers.");
+ }
+ return invokers.and(mockedInvokers);
}
}
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly Return. Reason: invocation.need.mock is set but not match true" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly Return. Reason: invocation.need.mock is set but not match true");
+ }
+ return invokers;
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockStateRouterFactory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockStateRouterFactory.java
index 8297700..4297ac9 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockStateRouterFactory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mock/MockStateRouterFactory.java
@@ -24,7 +24,7 @@ import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
/**
*
*/
-@Activate
+@Activate(order = -100)
public class MockStateRouterFactory implements StateRouterFactory {
public static final String NAME = "mock";
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
index ed4e413..3e2d570 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouter.java
@@ -19,14 +19,15 @@ package org.apache.dubbo.rpc.cluster.router.script;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
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.StateRouterResult;
import javax.script.Bindings;
import javax.script.Compilable;
@@ -50,7 +51,6 @@ import java.util.stream.Collectors;
import static org.apache.dubbo.rpc.cluster.Constants.DEFAULT_SCRIPT_TYPE_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.FORCE_KEY;
-import static org.apache.dubbo.rpc.cluster.Constants.PRIORITY_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.RULE_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.RUNTIME_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.TYPE_KEY;
@@ -85,7 +85,6 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
public ScriptStateRouter(URL url) {
super(url);
this.setUrl(url);
- this.setPriority(url.getParameter(PRIORITY_KEY, SCRIPT_ROUTER_DEFAULT_PRIORITY));
engine = getEngine(url);
rule = getRule(url);
@@ -124,16 +123,16 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
});
}
-
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url,
- Invocation invocation, boolean needToPrintMessage) throws RpcException {
+ protected BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder, Holder<String> messageHolder) throws RpcException {
if (engine == null || function == null) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly Return. Reason: engine or function is null" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly Return. Reason: engine or function is null");
+ }
+ return invokers;
}
Bindings bindings = createBindings(invokers, invocation);
- return new StateRouterResult<>(getRoutedInvokers(invokers, AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
+ return getRoutedInvokers(invokers, AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
try {
return function.eval(bindings);
} catch (ScriptException e) {
@@ -141,7 +140,7 @@ public class ScriptStateRouter<T> extends AbstractStateRouter<T> {
invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
return invokers;
}
- }, accessControlContext)));
+ }, accessControlContext));
}
/**
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
index 9f6e4a8..0c10d7a 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
@@ -17,23 +17,37 @@
package org.apache.dubbo.rpc.cluster.router.state;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.utils.Holder;
+import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.Constants;
import org.apache.dubbo.rpc.cluster.governance.GovernanceRuleRepository;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
+import org.apache.dubbo.rpc.model.ModuleModel;
/***
* The abstract class of StateRoute.
* @since 3.0
*/
public abstract class AbstractStateRouter<T> implements StateRouter<T> {
- private volatile int priority = DEFAULT_PRIORITY;
private volatile boolean force = false;
private volatile URL url;
+ private volatile StateRouter<T> nextRouter = null;
private final GovernanceRuleRepository ruleRepository;
+ /**
+ * Should continue route if current router's result is empty
+ */
+ private final boolean shouldFailFast;
+
public AbstractStateRouter(URL url) {
- this.ruleRepository = url.getOrDefaultModuleModel().getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension();
+ ModuleModel moduleModel = url.getOrDefaultModuleModel();
+ this.ruleRepository = moduleModel.getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension();
this.url = url;
+ this.shouldFailFast = Boolean.parseBoolean(ConfigurationUtils.getProperty(moduleModel, Constants.SHOULD_FAIL_FAST_KEY, "true"));
}
@Override
@@ -59,21 +73,117 @@ public abstract class AbstractStateRouter<T> implements StateRouter<T> {
this.force = force;
}
- @Override
- public int getPriority() {
- return priority;
+ public GovernanceRuleRepository getRuleRepository() {
+ return this.ruleRepository;
}
- public void setPriority(int priority) {
- this.priority = priority;
+ public StateRouter<T> getNextRouter() {
+ return nextRouter;
}
- public GovernanceRuleRepository getRuleRepository() {
- return this.ruleRepository;
+ @Override
+ public void notify(BitList<Invoker<T>> invokers) {
+ // default empty implement
}
@Override
- public void notify(BitList<Invoker<T>> invokers) {
+ public final BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder) throws RpcException {
+ if (needToPrintMessage && (nodeHolder == null || nodeHolder.get() == null)) {
+ needToPrintMessage = false;
+ }
+
+ RouterSnapshotNode<T> currentNode = null;
+ RouterSnapshotNode<T> parentNode = null;
+ Holder<String> messageHolder = null;
+
+ // pre-build current node
+ if (needToPrintMessage) {
+ parentNode = nodeHolder.get();
+ currentNode = new RouterSnapshotNode<>(this.getClass().getSimpleName(), invokers.clone());
+ parentNode.appendNode(currentNode);
+ // set parent node's output size in the first child invoke
+ // initial node output size is zero, first child will override it
+ if (parentNode.getNodeOutputSize() < invokers.size()) {
+ parentNode.setNodeOutputInvokers(invokers.clone());
+ }
+
+ messageHolder = new Holder<>();
+ nodeHolder.set(currentNode);
+ }
+ BitList<Invoker<T>> routeResult;
+
+ // check if router support call continue route by itself
+ if (!supportContinueRoute()) {
+ routeResult = doRoute(invokers, url, invocation, needToPrintMessage, nodeHolder, messageHolder);
+ // use current node's result as next node's parameter
+ if (!shouldFailFast || !routeResult.isEmpty()) {
+ routeResult = continueRoute(routeResult, url, invocation, needToPrintMessage, nodeHolder);
+ }
+ } else {
+ routeResult = doRoute(invokers, url, invocation, needToPrintMessage, nodeHolder, messageHolder);
+ }
+
+ // post-build current node
+ if (needToPrintMessage) {
+ currentNode.setRouterMessage(messageHolder.get());
+ if (currentNode.getNodeOutputSize() == 0) {
+ // no child call
+ currentNode.setNodeOutputInvokers(routeResult.clone());
+ }
+ currentNode.setChainOutputInvokers(routeResult.clone());
+ nodeHolder.set(parentNode);
+ }
+ return routeResult;
+ }
+
+ /**
+ * Filter invokers with current routing rule and only return the invokers that comply with the rule.
+ *
+ * @param invokers all invokers to be routed
+ * @param url consumerUrl
+ * @param invocation invocation
+ * @param needToPrintMessage should current router print message
+ * @param nodeHolder RouterSnapshotNode In general, router itself no need to care this param, just pass to continueRoute
+ * @param messageHolder message holder when router should current router print message
+ * @return routed result
+ */
+ protected abstract BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder,
+ Holder<String> messageHolder) throws RpcException;
+
+ /**
+ * Call next router to get result
+ *
+ * @param invokers current router filtered invokers
+ */
+ protected final BitList<Invoker<T>> continueRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder) {
+ if (nextRouter != null) {
+ return nextRouter.route(invokers, url, invocation, needToPrintMessage, nodeHolder);
+ } else {
+ return invokers;
+ }
+ }
+
+ /**
+ * Whether current router's implementation support call
+ * {@link AbstractStateRouter#continueRoute(BitList, URL, Invocation, boolean, Holder)}
+ * by router itself.
+ *
+ * @return support or not
+ */
+ protected boolean supportContinueRoute() {
+ return false;
+ }
+
+ /**
+ * Next Router node state is maintained by AbstractStateRouter and this method is not allow to override.
+ * If a specified router wants to control the behaviour of continue route or not,
+ * please override {@link AbstractStateRouter#supportContinueRoute()}
+ */
+ @Override
+ public final void setNextRouter(StateRouter<T> nextRouter) {
+ this.nextRouter = nextRouter;
}
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
index d26b7ff..bb4dd23 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java
@@ -21,6 +21,7 @@ import org.apache.dubbo.common.utils.CollectionUtils;
import java.util.AbstractList;
import java.util.ArrayList;
import java.util.BitSet;
+import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
@@ -63,12 +64,15 @@ public class BitList<E> extends AbstractList<E> {
public BitList(List<E> originList, boolean empty) {
if (originList instanceof BitList) {
this.originList = ((BitList<E>) originList).getOriginList();
+ this.tailList = ((BitList<E>) originList).getTailList();
} else {
this.originList = originList;
}
this.rootSet = new BitSet();
if (!empty) {
this.rootSet.set(0, originList.size());
+ } else {
+ this.tailList = null;
}
}
@@ -134,6 +138,13 @@ public class BitList<E> extends AbstractList<E> {
return tailList;
}
+ public void addToTailList(E e) {
+ if (tailList == null) {
+ tailList = new LinkedList<>();
+ }
+ tailList.add(e);
+ }
+
@SuppressWarnings("unchecked")
public static <T> BitList<T> emptyList() {
return emptyList;
@@ -159,7 +170,7 @@ public class BitList<E> extends AbstractList<E> {
/**
* If the element to added is appeared in originList even if it is not in rootSet,
* directly set its index in rootSet to true. (This may change the order of elements.)
- *
+ *
* If the element is not contained in originList, allocate tailList and add to tailList.
*
* Notice: It is not recommended adding duplicated element.
@@ -181,7 +192,7 @@ public class BitList<E> extends AbstractList<E> {
/**
* If the element to added is appeared in originList,
* directly set its index in rootSet to false. (This may change the order of elements.)
- *
+ *
* If the element is not contained in originList, try to remove from tailList.
*/
@Override
@@ -198,7 +209,7 @@ public class BitList<E> extends AbstractList<E> {
}
/**
- * Caution: This operation will clear originList for removing references purpose.
+ * Caution: This operation will clear originList for removing references purpose.
* This may change the default behaviour when adding new element later.
*/
@Override
@@ -270,6 +281,21 @@ public class BitList<E> extends AbstractList<E> {
}
@Override
+ @SuppressWarnings("unchecked")
+ public boolean addAll(Collection<? extends E> c) {
+ if (c instanceof BitList) {
+ rootSet.or(((BitList<? extends E>) c).rootSet);
+ if (((BitList<? extends E>) c).hasMoreElementInTailList()) {
+ for (E e : ((BitList<? extends E>) c).tailList) {
+ addToTailList(e);
+ }
+ }
+ return true;
+ }
+ return super.addAll(c);
+ }
+
+ @Override
public int lastIndexOf(Object o) {
int bitIndex = -1;
int index = -1;
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
index d84ac43..4d7820c 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouter.java
@@ -17,23 +17,26 @@
package org.apache.dubbo.rpc.cluster.router.state;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
/**
* State Router. (SPI, Prototype, ThreadSafe)
* <p>
* <a href="http://en.wikipedia.org/wiki/Routing">Routing</a>
*
+ * It is recommended to implement StateRouter by extending {@link AbstractStateRouter}
+ *
* @see org.apache.dubbo.rpc.cluster.Cluster#join(Directory, boolean)
+ * @see AbstractStateRouter
* @see Directory#list(Invocation)
* @since 3.0
*/
-public interface StateRouter<T> extends Comparable<StateRouter<T>> {
-
- int DEFAULT_PRIORITY = Integer.MAX_VALUE;
+public interface StateRouter<T> {
/**
* Get the router url.
@@ -43,7 +46,6 @@ public interface StateRouter<T> extends Comparable<StateRouter<T>> {
URL getUrl();
/***
- * ** This method can return the state of whether routerChain needed to continue route. **
* Filter invokers with current routing rule and only return the invokers that comply with the rule.
* Caching address lists in BitMap mode improves routing performance.
* @param invokers invoker bit list
@@ -53,8 +55,8 @@ public interface StateRouter<T> extends Comparable<StateRouter<T>> {
* @return state with route result
* @since 3.0
*/
- StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
- boolean needToPrintMessage) throws RpcException;
+ BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url, Invocation invocation,
+ boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder) throws RpcException;
/**
* To decide whether this router need to execute every time an RPC comes or should only execute when addresses or
@@ -66,7 +68,7 @@ public interface StateRouter<T> extends Comparable<StateRouter<T>> {
/**
* To decide whether this router should take effect when none of the invoker can match the router rule, which
- * means the {@link #route(BitList, URL, Invocation, boolean)} would be empty. Most of time, most router implementation would
+ * means the {@link #route(BitList, URL, Invocation, boolean, Holder)} would be empty. Most of time, most router implementation would
* default this value to false.
*
* @return true to execute if none of invokers matches the current router
@@ -74,29 +76,22 @@ public interface StateRouter<T> extends Comparable<StateRouter<T>> {
boolean isForce();
/**
- * Router's priority, used to sort routers.
- *
- * @return router's priority
- */
- int getPriority();
-
- /**
* 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 StateRouter#route(BitList, URL, Invocation, boolean)} gets called.
+ * chance to prepare before {@link StateRouter#route(BitList, URL, Invocation, boolean, Holder)} gets called.
+ * No need to notify next node.
*
* @param invokers invoker list
*/
void notify(BitList<Invoker<T>> invokers);
- @Override
- default int compareTo(StateRouter o) {
- if (o == null) {
- throw new IllegalArgumentException();
- }
- return Integer.compare(this.getPriority(), o.getPriority());
- }
-
default void stop() {
//do nothing by default
}
+
+ /**
+ * Notify next router node to current router.
+ *
+ * @param nextRouter next router node
+ */
+ void setNextRouter(StateRouter<T> nextRouter);
}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java
deleted file mode 100644
index b4fb316..0000000
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterResult.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.state;
-
-public class StateRouterResult<T> {
- private final boolean needContinueRoute;
- private final BitList<T> result;
- private final String message;
-
- public StateRouterResult(BitList<T> result) {
- this.needContinueRoute = true;
- this.result = result;
- this.message = null;
- }
-
- public StateRouterResult(BitList<T> invokers, BitList<T> routedResult) {
- this.needContinueRoute = true;
- this.result = invokers.and(routedResult);
- this.message = null;
- }
-
- public StateRouterResult(BitList<T> result, String message) {
- this.needContinueRoute = true;
- this.result = result;
- this.message = message;
- }
-
- public StateRouterResult(BitList<T> invokers, BitList<T> routedResult, String message) {
- this.needContinueRoute = true;
- this.result = invokers.and(routedResult);
- this.message = message;
- }
-
- public StateRouterResult(boolean needContinueRoute, BitList<T> result) {
- this.needContinueRoute = needContinueRoute;
- this.result = result;
- this.message = null;
- }
-
- public StateRouterResult(boolean needContinueRoute, BitList<T> invokers, BitList<T> routedResult) {
- this.needContinueRoute = needContinueRoute;
- this.result = invokers.and(routedResult);
- this.message = null;
- }
-
- public StateRouterResult(boolean needContinueRoute, BitList<T> result, String message) {
- this.needContinueRoute = needContinueRoute;
- this.result = result;
- this.message = message;
- }
-
- public StateRouterResult(boolean needContinueRoute, BitList<T> invokers, BitList<T> routedResult, String message) {
- this.needContinueRoute = needContinueRoute;
- this.result = invokers.and(routedResult);
- this.message = message;
- }
-
- public boolean isNeedContinueRoute() {
- return needContinueRoute;
- }
-
- public BitList<T> getResult() {
- return result;
- }
-
- public String getMessage() {
- return message;
- }
-}
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/TailStateRouter.java
similarity index 54%
copy from dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
copy to dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/TailStateRouter.java
index 9f6e4a8..d1cead7 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/TailStateRouter.java
@@ -17,59 +17,47 @@
package org.apache.dubbo.rpc.cluster.router.state;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
+import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.cluster.governance.GovernanceRuleRepository;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
-/***
- * The abstract class of StateRoute.
- * @since 3.0
- */
-public abstract class AbstractStateRouter<T> implements StateRouter<T> {
- private volatile int priority = DEFAULT_PRIORITY;
- private volatile boolean force = false;
- private volatile URL url;
-
- private final GovernanceRuleRepository ruleRepository;
+public class TailStateRouter<T> implements StateRouter<T> {
+ private static final TailStateRouter INSTANCE = new TailStateRouter();
- public AbstractStateRouter(URL url) {
- this.ruleRepository = url.getOrDefaultModuleModel().getExtensionLoader(GovernanceRuleRepository.class).getDefaultExtension();
- this.url = url;
+ @SuppressWarnings("unchecked")
+ public static <T> TailStateRouter<T> getInstance() {
+ return INSTANCE;
}
- @Override
- public URL getUrl() {
- return url;
- }
+ private TailStateRouter() {
- public void setUrl(URL url) {
- this.url = url;
}
@Override
- public boolean isRuntime() {
- return true;
- }
+ public void setNextRouter(StateRouter<T> nextRouter) {
- @Override
- public boolean isForce() {
- return force;
}
- public void setForce(boolean force) {
- this.force = force;
+ @Override
+ public URL getUrl() {
+ return null;
}
@Override
- public int getPriority() {
- return priority;
+ public BitList<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder) throws RpcException {
+ return invokers;
}
- public void setPriority(int priority) {
- this.priority = priority;
+ @Override
+ public boolean isRuntime() {
+ return false;
}
- public GovernanceRuleRepository getRuleRepository() {
- return this.ruleRepository;
+ @Override
+ public boolean isForce() {
+ return false;
}
@Override
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStateRouter.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStateRouter.java
index 3655955..5b57b21 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStateRouter.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/tag/TagStateRouter.java
@@ -24,14 +24,15 @@ import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
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.RouterSnapshotNode;
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.StateRouterResult;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRouterRule;
import org.apache.dubbo.rpc.cluster.router.tag.model.TagRuleParser;
@@ -47,7 +48,6 @@ import static org.apache.dubbo.rpc.Constants.FORCE_USE_TAG;
*/
public class TagStateRouter<T> extends AbstractStateRouter<T> implements ConfigurationListener {
public static final String NAME = "TAG_ROUTER";
- private static final int TAG_ROUTER_DEFAULT_PRIORITY = 100;
private static final Logger logger = LoggerFactory.getLogger(TagStateRouter.class);
private static final String RULE_SUFFIX = ".tag-router";
@@ -56,7 +56,6 @@ public class TagStateRouter<T> extends AbstractStateRouter<T> implements Configu
public TagStateRouter(URL url) {
super(url);
- this.setPriority(TAG_ROUTER_DEFAULT_PRIORITY);
}
@Override
@@ -79,18 +78,21 @@ public class TagStateRouter<T> extends AbstractStateRouter<T> implements Configu
}
@Override
- public StateRouterResult<Invoker<T>> route(BitList<Invoker<T>> invokers, URL url,
- Invocation invocation, boolean needToPrintMessage) throws RpcException {
+ public BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder, Holder<String> messageHolder) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
- return new StateRouterResult<>(invokers,
- needToPrintMessage ? "Directly Return. Reason: Invokers from previous router is empty." : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Directly Return. Reason: Invokers from previous router is empty.");
+ }
+ return invokers;
}
// since the rule can be changed by config center, we should copy one to use.
final TagRouterRule tagRouterRuleCopy = tagRouterRule;
if (tagRouterRuleCopy == null || !tagRouterRuleCopy.isValid() || !tagRouterRuleCopy.isEnabled()) {
- return new StateRouterResult<>(filterUsingStaticTag(invokers, url, invocation),
- needToPrintMessage ? "Disable Tag Router. Reason: tagRouterRule is invalid or disabled" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Disable Tag Router. Reason: tagRouterRule is invalid or disabled");
+ }
+ return filterUsingStaticTag(invokers, url, invocation);
}
BitList<Invoker<T>> result = invokers;
@@ -105,8 +107,10 @@ public class TagStateRouter<T> extends AbstractStateRouter<T> implements Configu
result = filterInvoker(invokers, invoker -> addressMatches(invoker.getUrl(), addresses));
// if result is not null OR it's null but force=true, return result directly
if (CollectionUtils.isNotEmpty(result) || tagRouterRuleCopy.isForce()) {
- return new StateRouterResult<>(result,
- needToPrintMessage ? "Use tag " + tag + " to route. Reason: result is not null OR it's null but force=true" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Use tag " + tag + " to route. Reason: result is not null OR it's null but force=true");
+ }
+ return result;
}
} else {
// dynamic tag group doesn't have any item about the requested app OR it's null after filtered by
@@ -116,15 +120,19 @@ public class TagStateRouter<T> extends AbstractStateRouter<T> implements Configu
// If there's no tagged providers that can match the current tagged request. force.tag is set by default
// to false, which means it will invoke any providers without a tag unless it's explicitly disallowed.
if (CollectionUtils.isNotEmpty(result) || isForceUseTag(invocation)) {
- return new StateRouterResult<>(result,
- needToPrintMessage ? "Use tag " + tag + " to route. Reason: result is not empty or ForceUseTag key is true in invocation" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("Use tag " + tag + " to route. Reason: result is not empty or ForceUseTag key is true in invocation");
+ }
+ return result;
}
// FAILOVER: return all Providers without any tags.
else {
BitList<Invoker<T>> tmp = filterInvoker(invokers, invoker -> addressNotMatches(invoker.getUrl(),
tagRouterRuleCopy.getAddresses()));
- return new StateRouterResult<>(filterInvoker(tmp, invoker -> StringUtils.isEmpty(invoker.getUrl().getParameter(TAG_KEY))),
- needToPrintMessage ? "FAILOVER: return all Providers without any tags" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("FAILOVER: return all Providers without any tags");
+ }
+ return filterInvoker(tmp, invoker -> StringUtils.isEmpty(invoker.getUrl().getParameter(TAG_KEY)));
}
} else {
// List<String> addresses = tagRouterRule.filter(providerApp);
@@ -134,16 +142,21 @@ public class TagStateRouter<T> extends AbstractStateRouter<T> implements Configu
result = filterInvoker(invokers, invoker -> addressNotMatches(invoker.getUrl(), addresses));
// 1. all addresses are in dynamic tag group, return empty list.
if (CollectionUtils.isEmpty(result)) {
- return new StateRouterResult<>(result,
- needToPrintMessage ? "all addresses are in dynamic tag group, return empty list" : null);
+ if (needToPrintMessage) {
+ messageHolder.set("all addresses are in dynamic tag group, return empty list");
+ }
+ return result;
}
// 2. if there are some addresses that are not in any dynamic tag group, continue to filter using the
// static tag group.
}
- return new StateRouterResult<>(filterInvoker(result, invoker -> {
+ if (needToPrintMessage) {
+ messageHolder.set("filter using the static tag group");
+ }
+ return filterInvoker(result, invoker -> {
String localTag = invoker.getUrl().getParameter(TAG_KEY);
return StringUtils.isEmpty(localTag) || !tagRouterRuleCopy.getTagNames().contains(localTag);
- }), needToPrintMessage ? "filter using the static tag group" : null);
+ });
}
}
diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
index 28ee7f1..90fbccd 100644
--- a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
+++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
@@ -2,6 +2,5 @@ tag=org.apache.dubbo.rpc.cluster.router.tag.TagStateRouterFactory
condition=org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouterFactory
service=org.apache.dubbo.rpc.cluster.router.condition.config.ServiceStateRouterFactory
app=org.apache.dubbo.rpc.cluster.router.condition.config.AppStateRouterFactory
-file=org.apache.dubbo.rpc.cluster.router.file.FileStateRouterFactory
mock=org.apache.dubbo.rpc.cluster.router.mock.MockStateRouterFactory
standard-mesh-rule=org.apache.dubbo.rpc.cluster.router.mesh.route.StandardMeshRuleRouterFactory
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
index e4a8f1a..288b7bc 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.directory;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
@@ -58,7 +59,7 @@ public class StaticDirectoryTest {
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
StaticDirectory<String> staticDirectory = new StaticDirectory<>(filteredInvokers);
boolean isAvailable = staticDirectory.isAvailable();
Assertions.assertTrue(!isAvailable);
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
index 3674b69..e7fa859 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionRouterTest.java
@@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.cluster.router.condition;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
@@ -123,12 +124,12 @@ public class ConditionRouterTest {
- List<Invoker<String>> filteredInvokers1 = router1.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
- List<Invoker<String>> filteredInvokers2 = router2.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
- List<Invoker<String>> filteredInvokers3 = router3.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
- List<Invoker<String>> filteredInvokers4 = router4.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
- List<Invoker<String>> filteredInvokers5 = router5.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
- List<Invoker<String>> filteredInvokers6 = router6.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers1 = router1.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
+ List<Invoker<String>> filteredInvokers2 = router2.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
+ List<Invoker<String>> filteredInvokers3 = router3.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
+ List<Invoker<String>> filteredInvokers4 = router4.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
+ List<Invoker<String>> filteredInvokers5 = router5.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
+ List<Invoker<String>> filteredInvokers6 = router6.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(1, filteredInvokers1.size());
Assertions.assertEquals(0, filteredInvokers2.size());
Assertions.assertEquals(0, filteredInvokers3.size());
@@ -177,14 +178,14 @@ public class ConditionRouterTest {
"host = " + LOCAL_HOST + " & methods = getFoo => " + " host = 10.20.3.3").addParameter(
FORCE_KEY, String.valueOf(true)));
List<Invoker<String>> filteredInvokers1 = router4.route(invokers,
- URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(1, filteredInvokers1.size());
StateRouter router5 = new ConditionStateRouterFactory().getRouter(String.class, getRouteUrl(
"host = " + LOCAL_HOST + " & methods = unvalidmethod => " + " host = 10.20.3.3")
.addParameter(FORCE_KEY, String.valueOf(true)));
List<Invoker<String>> filteredInvokers2 = router5.route(invokers,
- URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(3, filteredInvokers2.size());
// Request a non-exists method
}
@@ -198,7 +199,7 @@ public class ConditionRouterTest {
originInvokers.add(new MockInvoker<String>());
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -211,7 +212,7 @@ public class ConditionRouterTest {
originInvokers.add(new MockInvoker<String>());
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -224,7 +225,7 @@ public class ConditionRouterTest {
originInvokers.add(new MockInvoker<String>(URL.valueOf("dubbo://" + LOCAL_HOST + ":20880/com.foo.BarService")));
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -240,7 +241,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -258,7 +259,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -276,7 +277,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -294,7 +295,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -312,7 +313,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -328,7 +329,7 @@ public class ConditionRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(0, filteredInvokers.size());
}
@@ -347,25 +348,25 @@ public class ConditionRouterTest {
RpcInvocation invocation = new RpcInvocation();
String p = "a";
invocation.setArguments(new Object[]{null});
- List<Invoker<String>> fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ List<Invoker<String>> fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(3, fileredInvokers.size());
invocation.setArguments(new Object[]{p});
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(0, fileredInvokers.size());
router = new ConditionStateRouterFactory().getRouter(String.class, getRouteUrl("arguments = b " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(3, fileredInvokers.size());
router = new ConditionStateRouterFactory().getRouter(String.class, getRouteUrl("arguments[10].inner = a " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(3, fileredInvokers.size());
int integer = 1;
invocation.setArguments(new Object[]{integer});
router = new ConditionStateRouterFactory().getRouter(String.class, getRouteUrl("arguments[0].inner = 1 " + " => " + " host = 1.2.3.4").addParameter(FORCE_KEY, String.valueOf(true)));
- fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false).getResult();
+ fileredInvokers = router.route(invokers, URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"), invocation, false, new Holder<>());
Assertions.assertEquals(0, fileredInvokers.size());
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
index eb4419c..6cfc1e3 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
@@ -39,7 +39,6 @@ import org.junit.jupiter.api.Test;
import javax.script.ScriptEngineManager;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_CONNECTIVITY_VALIDATION;
@@ -173,7 +172,7 @@ public class FileRouterEngineTest {
URL dicInitUrl = URL.valueOf("consumer://localhost:20880/org.apache.dubbo.rpc.cluster.router.file.FileRouterEngineTest?application=FileRouterEngineTest");
dic = new StaticDirectory<>(dicInitUrl, invokers);
dic.buildRouterChain();
- dic.getRouterChain().initWithStateRouters(Collections.singletonList(routerFactory.getRouter(FileRouterEngineTest.class, url)));
+ dic.getRouterChain().setHeadStateRouter(routerFactory.getRouter(FileRouterEngineTest.class, url));
}
static class MockClusterInvoker<T> extends AbstractClusterInvoker<T> {
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
index feaff0b..02b86f7 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
@@ -20,6 +20,7 @@ package org.apache.dubbo.rpc.cluster.router.mesh.route;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
@@ -264,8 +265,10 @@ public class MeshRuleRouterTest {
public void testRoute1() {
StandardMeshRuleRouter<Object> meshRuleRouter = new StandardMeshRuleRouter<>(url);
BitList<Invoker<Object>> invokers = new BitList<>(Arrays.asList(createInvoker(""), createInvoker("unknown"), createInvoker("app1")));
- assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, null, false).getResult());
- assertEquals("MeshRuleCache has not been built. Skip route.", meshRuleRouter.route(invokers.clone(), null, null, true).getMessage());
+ assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, null, false, null));
+ Holder<String> message = new Holder<>();
+ meshRuleRouter.doRoute(invokers.clone(), null, null, true, null, message);
+ assertEquals("MeshRuleCache has not been built. Skip route.", message.get());
}
@Test
@@ -292,21 +295,25 @@ public class MeshRuleRouterTest {
rpcInvocation.setServiceName("ccc");
rpcInvocation.setAttachment("trafficLabel", "xxx");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(isolation, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
- assertEquals("Match App: app1 Subset: isolation ", meshRuleRouter.route(invokers.clone(), null, rpcInvocation, true).getMessage());
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(isolation, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
+ Holder<String> message = new Holder<>();
+ meshRuleRouter.doRoute(invokers.clone(), null, rpcInvocation, true, null, message);
+ assertEquals("Match App: app1 Subset: isolation ", message.get());
rpcInvocation.setAttachment("trafficLabel", "testing-trunk");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testingTrunk, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testingTrunk, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setAttachment("trafficLabel", null);
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setServiceName("aaa");
- assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult());
- assertEquals("Empty protection after routed.", meshRuleRouter.route(invokers.clone(), null, rpcInvocation, true).getMessage());
+ assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null));
+ message = new Holder<>();
+ meshRuleRouter.doRoute(invokers.clone(), null, rpcInvocation, true, null, message);
+ assertEquals("Empty protection after routed.", message.get());
rules = new LinkedList<>();
rules.add(yaml.load(rule1));
@@ -315,23 +322,23 @@ public class MeshRuleRouterTest {
rpcInvocation.setServiceName("ccc");
rpcInvocation.setAttachment("trafficLabel", "xxx");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(isolation, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(isolation, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setAttachment("trafficLabel", "testing-trunk");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testingTrunk, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testingTrunk, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setAttachment("trafficLabel", "testing");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setServiceName("aaa");
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
rpcInvocation.setAttachment("trafficLabel",null);
- assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult());
+ assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null));
rules = new LinkedList<>();
rules.add(yaml.load(rule1));
@@ -343,7 +350,7 @@ public class MeshRuleRouterTest {
int testingCount = 0;
int isolationCount = 0;
for (int i = 0; i < 1000; i++) {
- BitList<Invoker<Object>> result = meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult();
+ BitList<Invoker<Object>> result = meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null);
assertEquals(1, result.size());
if (result.contains(testing)) {
testingCount++;
@@ -355,15 +362,15 @@ public class MeshRuleRouterTest {
invokers.removeAll(Arrays.asList(isolation, testingTrunk));
for (int i = 0; i < 1000; i++) {
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
}
meshRuleRouter.notify(invokers);
for (int i = 0; i < 1000; i++) {
- assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().size());
- assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult().get(0));
+ assertEquals(1, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).size());
+ assertEquals(testing, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null).get(0));
}
Invoker<Object> mock = createInvoker(Collections.singletonMap("env-sign", "mock"));
@@ -371,7 +378,7 @@ public class MeshRuleRouterTest {
meshRuleRouter.notify(invokers);
invokers.removeAll(Arrays.asList(isolation, testingTrunk, testing));
- assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false).getResult());
+ assertEquals(invokers, meshRuleRouter.route(invokers.clone(), null, rpcInvocation, false, null));
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
index fcd0750..6798afd 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mock/MockInvokersSelectorTest.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.cluster.router.mock;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
@@ -53,7 +54,7 @@ public class MockInvokersSelectorTest {
selector.notify(providers);
// rpcInvocation does not have an attached "invocation.need.mock" parameter, so normal invokers will be filtered out
- List<Invoker<DemoService>> invokers = selector.route(providers.clone(), consumerURL, rpcInvocation, false).getResult();
+ List<Invoker<DemoService>> invokers = selector.route(providers.clone(), consumerURL, rpcInvocation, false, new Holder<>());
Assertions.assertEquals(invokers.size(),1);
Assertions.assertTrue(invokers.contains(invoker3));
@@ -61,7 +62,7 @@ public class MockInvokersSelectorTest {
Map<String,Object> attachments = new HashMap<>();
attachments.put(INVOCATION_NEED_MOCK,"true");
Mockito.when(rpcInvocation.getObjectAttachments()).thenReturn(attachments);
- invokers = selector.route(providers.clone(), consumerURL, rpcInvocation, false).getResult();
+ invokers = selector.route(providers.clone(), consumerURL, rpcInvocation, false, new Holder<>());
Assertions.assertEquals(invokers.size(),2);
Assertions.assertTrue(invokers.contains(invoker1));
Assertions.assertTrue(invokers.contains(invoker2));
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouterTest.java
index 4fe6dc5..8be79f9 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/script/ScriptStateRouterTest.java
@@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.cluster.router.script;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.router.MockInvoker;
@@ -62,7 +63,7 @@ public class ScriptStateRouterTest {
originInvokers.add(new MockInvoker<String>());
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(invokers, filteredInvokers);
}
@@ -87,7 +88,7 @@ public class ScriptStateRouterTest {
originInvokers.add(invoker3);
BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
- List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> filteredInvokers = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(2, filteredInvokers.size());
Assertions.assertEquals(invoker2, filteredInvokers.get(0));
Assertions.assertEquals(invoker3, filteredInvokers.get(1));
@@ -118,7 +119,7 @@ public class ScriptStateRouterTest {
"route(invokers, invocation, context) ";
StateRouter router = new ScriptStateRouterFactory().getRouter(String.class, getRouteUrl(script));
- List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(1, routeResult.size());
Assertions.assertEquals(invoker2, routeResult.get(0));
}
@@ -136,7 +137,7 @@ public class ScriptStateRouterTest {
String script = "/";
StateRouter router = new ScriptStateRouterFactory().getRouter(String.class, getRouteUrl(script));
- List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false).getResult();
+ List<Invoker<String>> routeResult = router.route(invokers, invokers.get(0).getUrl(), new RpcInvocation(), false, new Holder<>());
Assertions.assertEquals(3, routeResult.size());
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
index 76c98ae..c6e0117 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java
@@ -138,6 +138,26 @@ public class BitListTest {
}
@Test
+ public void testAddAll() {
+ List<String> list = Arrays.asList("A", "B", "C");
+ BitList<String> bitList1 = new BitList<>(list);
+ BitList<String> bitList2 = new BitList<>(list);
+
+ bitList1.removeAll(list);
+ Assertions.assertEquals(0, bitList1.size());
+
+ bitList1.addAll(bitList2);
+
+ Assertions.assertEquals(3, bitList1.size());
+ Assertions.assertFalse(bitList1.hasMoreElementInTailList());
+
+
+ bitList1.addAll(bitList2);
+
+ Assertions.assertEquals(3, bitList1.size());
+ }
+
+ @Test
public void testGet() {
List<String> list = Arrays.asList("A", "B", "C");
BitList<String> bitList = new BitList<>(list);
diff --git a/dubbo-cluster/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory b/dubbo-cluster/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
index fe1043e..8e551ae 100644
--- a/dubbo-cluster/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
+++ b/dubbo-cluster/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory
@@ -1 +1,2 @@
script=org.apache.dubbo.rpc.cluster.router.script.ScriptStateRouterFactory
+file=org.apache.dubbo.rpc.cluster.router.file.FileStateRouterFactory
diff --git a/dubbo-native/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory$Adaptive.java b/dubbo-native/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory$Adaptive.java
index afd4284..2dc6ccf 100644
--- a/dubbo-native/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory$Adaptive.java
+++ b/dubbo-native/src/main/java/org/apache/dubbo/rpc/cluster/router/state/StateRouterFactory$Adaptive.java
@@ -15,10 +15,11 @@
* limitations under the License.
*/
package org.apache.dubbo.rpc.cluster.router.state;
+import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.model.ScopeModel;
import org.apache.dubbo.rpc.model.ScopeModelUtil;
public class StateRouterFactory$Adaptive implements org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory {
-public org.apache.dubbo.rpc.cluster.router.state.StateRouter getRouter(java.lang.Class arg0, org.apache.dubbo.common.URL arg1) {
+public <T> org.apache.dubbo.rpc.cluster.router.state.StateRouter<T> getRouter(Class<T> arg0, URL arg1) {
if (arg0 == null) throw new IllegalArgumentException("url == null");
org.apache.dubbo.common.URL url = arg1;
String extName = ( url.getProtocol() == null ? "adaptive" : url.getProtocol() );
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java
index e59ae9b..4a7a1d0 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcServiceContext.java
@@ -73,6 +73,8 @@ public class RpcServiceContext extends RpcContext {
private Object response;
private AsyncContext asyncContext;
+ private boolean needPrintRouterSnapshot;
+
/**
* Get the request object of the underlying RPC protocol, e.g. HttpServletRequest
*
@@ -650,6 +652,14 @@ public class RpcServiceContext extends RpcContext {
rpcContext.setConsumerUrl(url);
}
+ public boolean isNeedPrintRouterSnapshot() {
+ return needPrintRouterSnapshot;
+ }
+
+ public void setNeedPrintRouterSnapshot(boolean needPrintRouterSnapshot) {
+ this.needPrintRouterSnapshot = needPrintRouterSnapshot;
+ }
+
/**
* Only part of the properties are copied, the others are either not used currently or can be got from invocation.
* Also see {@link RpcContextAttachment#copyOf(boolean)}