This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch 3.1
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.1 by this push:
new 8dec7d411d Fix wildcard match (#10256)
8dec7d411d is described below
commit 8dec7d411d0f045537516f3db2f916fd11486a72
Author: Albumen Kevin <[email protected]>
AuthorDate: Thu Jul 14 17:43:26 2022 +0800
Fix wildcard match (#10256)
---
.../apache/dubbo/common/ProtocolServiceKey.java | 124 ++++++++++++
.../java/org/apache/dubbo/common/ServiceKey.java | 146 +++++++++++++
.../common/ProtocolServiceKeyMatcherTest.java | 107 ++++++++++
.../dubbo/common/ProtocolServiceKeyTest.java | 76 +++++++
.../apache/dubbo/common/ServiceKeyMatcherTest.java | 224 ++++++++++++++++++++
.../org/apache/dubbo/common/ServiceKeyTest.java | 54 +++++
.../org/apache/dubbo/metadata/MetadataInfo.java | 29 ++-
.../registry/client/DefaultServiceInstance.java | 6 +
.../dubbo/registry/client/InstanceAddressURL.java | 21 ++
.../registry/client/ServiceDiscoveryRegistry.java | 8 +-
.../client/ServiceDiscoveryRegistryDirectory.java | 208 ++++++++++++++++---
.../listener/ServiceInstancesChangedListener.java | 213 ++++++++-----------
.../client/ServiceDiscoveryRegistryTest.java | 10 +-
.../MockServiceInstancesChangedListener.java | 5 +-
.../ServiceInstancesChangedListenerTest.java | 225 +++++++++++++++------
15 files changed, 1212 insertions(+), 244 deletions(-)
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/ProtocolServiceKey.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/ProtocolServiceKey.java
new file mode 100644
index 0000000000..0cc8b203d4
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/ProtocolServiceKey.java
@@ -0,0 +1,124 @@
+/*
+ * 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.common;
+
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.StringUtils;
+
+import java.util.Objects;
+
+public class ProtocolServiceKey extends ServiceKey {
+ private final String protocol;
+
+ public ProtocolServiceKey(String interfaceName, String version, String
group, String protocol) {
+ super(interfaceName, version, group);
+ this.protocol = protocol;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public String getServiceKeyString() {
+ return super.toString();
+ }
+
+ public boolean isSameWith(ProtocolServiceKey protocolServiceKey) {
+ // interface version group should be the same
+ if (!super.equals(protocolServiceKey)) {
+ return false;
+ }
+
+ // origin protocol is *, can not match any protocol
+ if (CommonConstants.ANY_VALUE.equals(protocol)) {
+ return false;
+ }
+
+ // origin protocol is null, can match any protocol
+ if (StringUtils.isEmpty(protocol) ||
StringUtils.isEmpty(protocolServiceKey.getProtocol())) {
+ return true;
+ }
+
+ // origin protocol is not *, match itself
+ return Objects.equals(protocol, protocolServiceKey.getProtocol());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ ProtocolServiceKey that = (ProtocolServiceKey) o;
+ return Objects.equals(protocol, that.protocol);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), protocol);
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + CommonConstants.GROUP_CHAR_SEPARATOR +
protocol;
+ }
+
+ public static class Matcher {
+ public static boolean isMatch(ProtocolServiceKey rule,
ProtocolServiceKey target) {
+ // 1. 2. 3. match interface / version / group
+ if (!ServiceKey.Matcher.isMatch(rule, target)) {
+ return false;
+ }
+
+ // 4.match protocol
+ // 4.1. if rule group is *, match all
+ if (!CommonConstants.ANY_VALUE.equals(rule.getProtocol())) {
+ // 4.2. if rule protocol is null, match all
+ if (StringUtils.isNotEmpty(rule.getProtocol())) {
+ // 4.3. if rule protocol contains ',', split and match each
+ if
(rule.getProtocol().contains(CommonConstants.COMMA_SEPARATOR)) {
+ String[] protocols = rule.getProtocol().split("\\"
+CommonConstants.COMMA_SEPARATOR, -1);
+ boolean match = false;
+ for (String protocol : protocols) {
+ protocol = protocol.trim();
+ if (StringUtils.isEmpty(protocol) &&
StringUtils.isEmpty(target.getProtocol())) {
+ match = true;
+ break;
+ } else if (protocol.equals(target.getProtocol())) {
+ match = true;
+ break;
+ }
+ }
+ if (!match) {
+ return false;
+ }
+ }
+ // 4.3. if rule protocol is not contains ',', match
directly
+ else if (!Objects.equals(rule.getProtocol(),
target.getProtocol())) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+ }
+}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/ServiceKey.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/ServiceKey.java
new file mode 100644
index 0000000000..c8d8f846cc
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/ServiceKey.java
@@ -0,0 +1,146 @@
+/*
+ * 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.common;
+
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.StringUtils;
+
+import java.util.Objects;
+
+public class ServiceKey {
+ private final String interfaceName;
+ private final String group;
+ private final String version;
+
+ public ServiceKey(String interfaceName, String version, String group) {
+ this.interfaceName = interfaceName;
+ this.group = group;
+ this.version = version;
+ }
+
+ public String getInterfaceName() {
+ return interfaceName;
+ }
+
+ public String getGroup() {
+ return group;
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ServiceKey that = (ServiceKey) o;
+ return Objects.equals(interfaceName, that.interfaceName) &&
Objects.equals(group, that.group) && Objects.equals(version, that.version);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(interfaceName, group, version);
+ }
+
+ @Override
+ public String toString() {
+ return BaseServiceMetadata.buildServiceKey(interfaceName, group,
version);
+ }
+
+
+ public static class Matcher {
+ public static boolean isMatch(ServiceKey rule, ServiceKey target) {
+ // 1. match interface (accurate match)
+ if (!Objects.equals(rule.getInterfaceName(),
target.getInterfaceName())) {
+ return false;
+ }
+
+ // 2. match version (accurate match)
+ // 2.1. if rule version is *, match all
+ if (!CommonConstants.ANY_VALUE.equals(rule.getVersion())) {
+ // 2.2. if rule version is null, target version should be null
+ if (StringUtils.isEmpty(rule.getVersion()) &&
!StringUtils.isEmpty(target.getVersion())) {
+ return false;
+ }
+ if (!StringUtils.isEmpty(rule.getVersion())) {
+ // 2.3. if rule version contains ',', split and match each
+ if
(rule.getVersion().contains(CommonConstants.COMMA_SEPARATOR)) {
+ String[] versions = rule.getVersion().split("\\"
+CommonConstants.COMMA_SEPARATOR, -1);
+ boolean match = false;
+ for (String version : versions) {
+ version = version.trim();
+ if (StringUtils.isEmpty(version) &&
StringUtils.isEmpty(target.getVersion())) {
+ match = true;
+ break;
+ } else if (version.equals(target.getVersion())) {
+ match = true;
+ break;
+ }
+ }
+ if (!match) {
+ return false;
+ }
+ }
+ // 2.4. if rule version is not contains ',', match directly
+ else if (!Objects.equals(rule.getVersion(),
target.getVersion())) {
+ return false;
+ }
+ }
+ }
+
+ // 3. match group (wildcard match)
+ // 3.1. if rule group is *, match all
+ if (!CommonConstants.ANY_VALUE.equals(rule.getGroup())) {
+ // 3.2. if rule group is null, target group should be null
+ if (StringUtils.isEmpty(rule.getGroup()) &&
!StringUtils.isEmpty(target.getGroup())) {
+ return false;
+ }
+ if (!StringUtils.isEmpty(rule.getGroup())) {
+ // 3.3. if rule group contains ',', split and match each
+ if
(rule.getGroup().contains(CommonConstants.COMMA_SEPARATOR)) {
+ String[] groups = rule.getGroup().split("\\"
+CommonConstants.COMMA_SEPARATOR, -1);
+ boolean match = false;
+ for (String group : groups) {
+ group = group.trim();
+ if (StringUtils.isEmpty(group) &&
StringUtils.isEmpty(target.getGroup())) {
+ match = true;
+ break;
+ } else if (group.equals(target.getGroup())) {
+ match = true;
+ break;
+ }
+ }
+ if (!match) {
+ return false;
+ }
+ }
+ // 3.4. if rule group is not contains ',', match directly
+ else if (!Objects.equals(rule.getGroup(),
target.getGroup())) {
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+ }
+}
diff --git
a/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyMatcherTest.java
b/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyMatcherTest.java
new file mode 100644
index 0000000000..635045c408
--- /dev/null
+++
b/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyMatcherTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.common;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ProtocolServiceKeyMatcherTest {
+
+ @Test
+ public void testProtocol() {
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo"),
+ new ProtocolServiceKey(null, null, null, null)
+ ));
+
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo"),
+ new ProtocolServiceKey("DemoService", null, null, "dubbo")
+ ));
+
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, null),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, ""),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "*"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "dubbo1")
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "dubbo2")
+ ));
+
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,,dubbo2"),
+ new ProtocolServiceKey(null, null, null, null)
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "")
+ ));
+
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, ",dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, null)
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, ",dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "")
+ ));
+
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2,"),
+ new ProtocolServiceKey(null, null, null, null)
+ ));
+ Assertions.assertTrue(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2,"),
+ new ProtocolServiceKey(null, null, null, "")
+ ));
+
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, ",dubbo1,dubbo2"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ Assertions.assertFalse(ProtocolServiceKey.Matcher.isMatch(
+ new ProtocolServiceKey(null, null, null, "dubbo1,dubbo2,"),
+ new ProtocolServiceKey(null, null, null, "dubbo")
+ ));
+ }
+}
diff --git
a/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyTest.java
b/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyTest.java
new file mode 100644
index 0000000000..09b2711aa2
--- /dev/null
+++
b/dubbo-common/src/test/java/org/apache/dubbo/common/ProtocolServiceKeyTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.common;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ProtocolServiceKeyTest {
+ @Test
+ public void test() {
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol1");
+ Assertions.assertEquals("DemoService",
protocolServiceKey.getInterfaceName());
+ Assertions.assertEquals("1.0.0", protocolServiceKey.getVersion());
+ Assertions.assertEquals("group1", protocolServiceKey.getGroup());
+ Assertions.assertEquals("protocol1", protocolServiceKey.getProtocol());
+
+ Assertions.assertEquals("group1/DemoService:1.0.0:protocol1",
protocolServiceKey.toString());
+ Assertions.assertEquals("group1/DemoService:1.0.0",
protocolServiceKey.getServiceKeyString());
+
+ Assertions.assertEquals(protocolServiceKey, protocolServiceKey);
+
+ ProtocolServiceKey protocolServiceKey1 = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol1");
+ Assertions.assertEquals(protocolServiceKey, protocolServiceKey1);
+ Assertions.assertEquals(protocolServiceKey.hashCode(),
protocolServiceKey1.hashCode());
+
+ ProtocolServiceKey protocolServiceKey2 = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol2");
+ Assertions.assertNotEquals(protocolServiceKey, protocolServiceKey2);
+
+ ProtocolServiceKey protocolServiceKey3 = new
ProtocolServiceKey("DemoService", "1.0.0", "group2", "protocol1");
+ Assertions.assertNotEquals(protocolServiceKey, protocolServiceKey3);
+
+ ProtocolServiceKey protocolServiceKey4 = new
ProtocolServiceKey("DemoService", "1.0.1", "group1", "protocol1");
+ Assertions.assertNotEquals(protocolServiceKey, protocolServiceKey4);
+
+ ProtocolServiceKey protocolServiceKey5 = new
ProtocolServiceKey("DemoInterface", "1.0.0", "group1", "protocol1");
+ Assertions.assertNotEquals(protocolServiceKey, protocolServiceKey5);
+
+ ServiceKey serviceKey = new ServiceKey("DemoService", "1.0.0",
"group1");
+ Assertions.assertNotEquals(protocolServiceKey, serviceKey);
+
+
Assertions.assertTrue(protocolServiceKey.isSameWith(protocolServiceKey));
+ Assertions.assertTrue(protocolServiceKey.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "")));
+ Assertions.assertTrue(protocolServiceKey.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", null)));
+
+ Assertions.assertFalse(protocolServiceKey.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group2", "protocol1")));
+ Assertions.assertFalse(protocolServiceKey.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group2", "")));
+ Assertions.assertFalse(protocolServiceKey.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group2", null)));
+
+
+ ProtocolServiceKey protocolServiceKey6 = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", null);
+
Assertions.assertTrue(protocolServiceKey6.isSameWith(protocolServiceKey6));
+ Assertions.assertTrue(protocolServiceKey6.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "")));
+ Assertions.assertTrue(protocolServiceKey6.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol1")));
+ Assertions.assertTrue(protocolServiceKey6.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol2")));
+
+ ProtocolServiceKey protocolServiceKey7 = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "*");
+ Assertions.assertFalse(protocolServiceKey7.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", null)));
+ Assertions.assertFalse(protocolServiceKey7.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "")));
+ Assertions.assertFalse(protocolServiceKey7.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol1")));
+ Assertions.assertFalse(protocolServiceKey7.isSameWith(new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol2")));
+ }
+}
diff --git
a/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyMatcherTest.java
b/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyMatcherTest.java
new file mode 100644
index 0000000000..6d85c46b31
--- /dev/null
+++
b/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyMatcherTest.java
@@ -0,0 +1,224 @@
+/*
+ * 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.common;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ServiceKeyMatcherTest {
+
+ @Test
+ public void testInterface() {
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey("DemoService", null, null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, null),
+ new ServiceKey("DemoService", null, null)
+ ));
+
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey("*", null, null),
+ new ServiceKey("DemoService", null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey("*", null, null),
+ new ServiceKey(null, null, null)
+ ));
+ }
+
+ @Test
+ public void testVersion() {
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0", null),
+ new ServiceKey(null, "1.0.0", null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, null),
+ new ServiceKey(null, "1.0.0", null)
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1", null),
+ new ServiceKey(null, "1.0.0", null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1", null),
+ new ServiceKey(null, "1.0.2", null)
+ ));
+
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1", null),
+ new ServiceKey(null, null, null)
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, ",1.0.0,1.0.1", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, ",1.0.0,1.0.1", null),
+ new ServiceKey(null, "", null)
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,,1.0.1", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,,1.0.1", null),
+ new ServiceKey(null, "", null)
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1,", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1,", null),
+ new ServiceKey(null, "", null)
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "1.0.0,1.0.1", null),
+ new ServiceKey(null, "", null)
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, ",1.0.0,1.0.1", null),
+ new ServiceKey(null, "1.0.2", null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, ",1.0.0,1.0.1", null),
+ new ServiceKey(null, "1.0.2", null)
+ ));
+
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "*", null),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "*", null),
+ new ServiceKey(null, "", null)
+ ));
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, "*", null),
+ new ServiceKey(null, "1.0.0", null)
+ ));
+ }
+
+ @Test
+ public void testGroup() {
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group1"),
+ new ServiceKey(null, null, "group1")
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group1"),
+ new ServiceKey(null, null, null)
+ ));
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, null),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group1, group2"),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2, group3"),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2, group3"),
+ new ServiceKey(null, null, null)
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2, group3"),
+ new ServiceKey(null, null, "")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, ",group2"),
+ new ServiceKey(null, null, "")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2,"),
+ new ServiceKey(null, null, "")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2, ,group3"),
+ new ServiceKey(null, null, "")
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, ",group2"),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2,"),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertFalse(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "group2, ,group3"),
+ new ServiceKey(null, null, "group1")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "*"),
+ new ServiceKey(null, null, "")
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "*"),
+ new ServiceKey(null, null, null)
+ ));
+
+ Assertions.assertTrue(ServiceKey.Matcher.isMatch(
+ new ServiceKey(null, null, "*"),
+ new ServiceKey(null, null, "group1")
+ ));
+ }
+}
diff --git
a/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyTest.java
b/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyTest.java
new file mode 100644
index 0000000000..5643166ffa
--- /dev/null
+++ b/dubbo-common/src/test/java/org/apache/dubbo/common/ServiceKeyTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.common;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ServiceKeyTest {
+ @Test
+ public void test() {
+ ServiceKey serviceKey = new ServiceKey("DemoService", "1.0.0",
"group1");
+
+ Assertions.assertEquals("DemoService", serviceKey.getInterfaceName());
+ Assertions.assertEquals("1.0.0", serviceKey.getVersion());
+ Assertions.assertEquals("group1", serviceKey.getGroup());
+
+ Assertions.assertEquals("group1/DemoService:1.0.0",
serviceKey.toString());
+ Assertions.assertEquals("DemoService", new ServiceKey("DemoService",
null, null).toString());
+ Assertions.assertEquals("DemoService:1.0.0", new
ServiceKey("DemoService", "1.0.0", null).toString());
+ Assertions.assertEquals("group1/DemoService", new
ServiceKey("DemoService", null, "group1").toString());
+
+ Assertions.assertEquals(serviceKey, serviceKey);
+
+ ServiceKey serviceKey1 = new ServiceKey("DemoService", "1.0.0",
"group1");
+ Assertions.assertEquals(serviceKey, serviceKey1);
+ Assertions.assertEquals(serviceKey.hashCode(), serviceKey1.hashCode());
+
+ ServiceKey serviceKey2 = new ServiceKey("DemoService", "1.0.0",
"group2");
+ Assertions.assertNotEquals(serviceKey, serviceKey2);
+
+ ServiceKey serviceKey3 = new ServiceKey("DemoService", "1.0.1",
"group1");
+ Assertions.assertNotEquals(serviceKey, serviceKey3);
+
+ ServiceKey serviceKey4 = new ServiceKey("DemoInterface", "1.0.0",
"group1");
+ Assertions.assertNotEquals(serviceKey, serviceKey4);
+
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey("DemoService", "1.0.0", "group1", "protocol1");
+ Assertions.assertNotEquals(serviceKey, protocolServiceKey);
+ }
+}
diff --git
a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataInfo.java
b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataInfo.java
index 2735b931ee..0465fbb228 100644
---
a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataInfo.java
+++
b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataInfo.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.metadata;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
@@ -437,6 +438,7 @@ public class MetadataInfo implements Serializable {
private String group;
private String version;
private String protocol;
+ private int port = -1;
private String path; // most of the time, path is the same with the
interface name.
private Map<String, String> params;
@@ -453,12 +455,14 @@ public class MetadataInfo implements Serializable {
// service + group + version + protocol
private volatile transient String matchKey;
+ private volatile transient ProtocolServiceKey protocolServiceKey;
+
private transient URL url;
public ServiceInfo() {}
public ServiceInfo(URL url, List<MetadataParamsFilter> filters) {
- this(url.getServiceInterface(), url.getGroup(), url.getVersion(),
url.getProtocol(), url.getPath(), null);
+ this(url.getServiceInterface(), url.getGroup(), url.getVersion(),
url.getProtocol(), url.getPort(), url.getPath(), null);
this.url = url;
Map<String, String> params = extractServiceParams(url, filters);
// initialize method params caches.
@@ -466,11 +470,12 @@ public class MetadataInfo implements Serializable {
this.consumerMethodParams =
URLParam.initMethodParameters(consumerParams);
}
- public ServiceInfo(String name, String group, String version, String
protocol, String path, Map<String, String> params) {
+ public ServiceInfo(String name, String group, String version, String
protocol, int port, String path, Map<String, String> params) {
this.name = name;
this.group = group;
this.version = version;
this.protocol = protocol;
+ this.port = port;
this.path = path;
this.params = params == null ? new ConcurrentHashMap<>() : params;
@@ -577,6 +582,14 @@ public class MetadataInfo implements Serializable {
return matchKey;
}
+ public ProtocolServiceKey getProtocolServiceKey() {
+ if (protocolServiceKey != null) {
+ return protocolServiceKey;
+ }
+ protocolServiceKey = new ProtocolServiceKey(name, version, group,
protocol);
+ return protocolServiceKey;
+ }
+
private String buildServiceKey(String name, String group, String
version) {
this.serviceKey = URL.buildKey(name, group, version);
return this.serviceKey;
@@ -630,6 +643,14 @@ public class MetadataInfo implements Serializable {
this.protocol = protocol;
}
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
public Map<String, String> getParams() {
if (params == null) {
return Collections.emptyMap();
@@ -759,12 +780,13 @@ public class MetadataInfo implements Serializable {
&& Objects.equals(this.getGroup(), serviceInfo.getGroup())
&& Objects.equals(this.getName(), serviceInfo.getName())
&& Objects.equals(this.getProtocol(),
serviceInfo.getProtocol())
+ && Objects.equals(this.getPort(), serviceInfo.getPort())
&& this.getParams().equals(serviceInfo.getParams());
}
@Override
public int hashCode() {
- return Objects.hash(getVersion(), getGroup(), getName(),
getProtocol(), getParams());
+ return Objects.hash(getVersion(), getGroup(), getName(),
getProtocol(), getPort(), getParams());
}
@Override
@@ -778,6 +800,7 @@ public class MetadataInfo implements Serializable {
"group='" + group + "'," +
"version='" + version + "'," +
"protocol='" + protocol + "'," +
+ "port='" + port + "'," +
"params=" + params + "," +
"}";
}
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceInstance.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceInstance.java
index 074811beed..432a77738a 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceInstance.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceInstance.java
@@ -245,6 +245,12 @@ public class DefaultServiceInstance implements
ServiceInstance {
return copyOfInstance;
}
+ public DefaultServiceInstance copyFrom(int port) {
+ DefaultServiceInstance copyOfInstance = new
DefaultServiceInstance(this);
+ copyOfInstance.setPort(port);
+ return copyOfInstance;
+ }
+
@Override
public Map<String, String> getAllParams() {
if (extendParams == null) {
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java
index 8e8b2f9424..225b6e6e2e 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/InstanceAddressURL.java
@@ -17,6 +17,7 @@
package org.apache.dubbo.registry.client;
import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.url.component.URLAddress;
import org.apache.dubbo.common.url.component.URLParam;
import org.apache.dubbo.common.utils.CollectionUtils;
@@ -126,6 +127,26 @@ public class InstanceAddressURL extends URL {
return RpcContext.getServiceContext().getServiceKey();
}
+ @Override
+ public URL setProtocol(String protocol) {
+ return new ServiceConfigURL(protocol, getUsername(), getPassword(),
getHost(), getPort(), getPath(), getParameters(), attributes);
+ }
+
+ @Override
+ public URL setHost(String host) {
+ return new ServiceConfigURL(getProtocol(), getUsername(),
getPassword(), host, getPort(), getPath(), getParameters(), attributes);
+ }
+
+ @Override
+ public URL setPort(int port) {
+ return new ServiceConfigURL(getProtocol(), getUsername(),
getPassword(), getHost(), port, getPath(), getParameters(), attributes);
+ }
+
+ @Override
+ public URL setPath(String path) {
+ return new ServiceConfigURL(getProtocol(), getUsername(),
getPassword(), getHost(), getPort(), path, getParameters(), attributes);
+ }
+
@Override
public String getAddress() {
return instance.getAddress();
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
index 30eb983edd..a1aaf01cfe 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
@@ -299,8 +299,8 @@ public class ServiceDiscoveryRegistry extends
FailbackRegistry {
protected void subscribeURLs(URL url, NotifyListener listener, Set<String>
serviceNames) {
serviceNames = toTreeSet(serviceNames);
String serviceNamesKey = toStringKeys(serviceNames);
- String protocolServiceKey = url.getProtocolServiceKey();
- logger.info(String.format("Trying to subscribe from apps %s for
service key %s, ", serviceNamesKey, protocolServiceKey));
+ String serviceKey = url.getServiceKey();
+ logger.info(String.format("Trying to subscribe from apps %s for
service key %s, ", serviceNamesKey, serviceKey));
// register ServiceInstancesChangedListener
Lock appSubscriptionLock = getAppSubscription(serviceNamesKey);
@@ -322,7 +322,7 @@ public class ServiceDiscoveryRegistry extends
FailbackRegistry {
if (!serviceInstancesChangedListener.isDestroyed()) {
serviceInstancesChangedListener.setUrl(url);
listener.addServiceListener(serviceInstancesChangedListener);
-
serviceInstancesChangedListener.addListenerAndNotify(protocolServiceKey,
listener);
+ serviceInstancesChangedListener.addListenerAndNotify(url,
listener);
serviceDiscovery.addServiceInstancesChangedListener(serviceInstancesChangedListener);
} else {
logger.info(String.format("Listener of %s has been destroyed
by another thread.", serviceNamesKey));
@@ -398,7 +398,7 @@ public class ServiceDiscoveryRegistry extends
FailbackRegistry {
Lock appSubscriptionLock =
getAppSubscription(appKey);
try {
appSubscriptionLock.lock();
-
oldListener.removeListener(url.getProtocolServiceKey(), listener);
+
oldListener.removeListener(url.getServiceKey(), listener);
if (!oldListener.hasListeners()) {
oldListener.destroy();
removeAppSubscriptionLock(appKey);
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
index bfe1b8b383..7bb611e3d3 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
@@ -16,8 +16,10 @@
*/
package org.apache.dubbo.registry.client;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
+import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
@@ -30,26 +32,35 @@ import org.apache.dubbo.registry.Constants;
import org.apache.dubbo.registry.ProviderFirstParams;
import org.apache.dubbo.registry.integration.AbstractConfiguratorListener;
import org.apache.dubbo.registry.integration.DynamicDirectory;
+import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
+import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.cluster.Configurator;
import org.apache.dubbo.rpc.cluster.RouterChain;
+import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import static org.apache.dubbo.common.constants.CommonConstants.DISABLED_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.ENABLED_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static
org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_HASHMAP_LOAD_FACTOR;
import static
org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL;
import static
org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_TYPE_KEY;
@@ -64,13 +75,15 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
* instance address to invoker mapping.
* The initial value is null and the midway may be assigned to null,
please use the local variable reference
*/
- private volatile Map<String, Invoker<T>> urlInvokerMap;
+ private volatile Map<ProtocolServiceKeyWithAddress, Invoker<T>>
urlInvokerMap;
private volatile ReferenceConfigurationListener
referenceConfigurationListener;
private volatile boolean enableConfigurationListen = true;
private volatile List<URL> originalUrls = null;
private volatile Map<String, String> overrideQueryMap;
private final Set<String> providerFirstParams;
private final ModuleModel moduleModel;
+ private final ProtocolServiceKey consumerProtocolServiceKey;
+ private final Map<ProtocolServiceKey, URL> customizedConsumerUrlMap = new
ConcurrentHashMap<>();
public ServiceDiscoveryRegistryDirectory(Class<T> serviceType, URL url) {
super(serviceType, url);
@@ -94,6 +107,9 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
}
}
+ String protocol = consumerUrl.getParameter(PROTOCOL_KEY,
consumerUrl.getProtocol());
+ consumerProtocolServiceKey = new
ProtocolServiceKey(consumerUrl.getServiceInterface(), consumerUrl.getVersion(),
consumerUrl.getGroup(),
+ !CommonConstants.CONSUMER.equals(protocol) ? protocol : null);
}
@Override
@@ -224,15 +240,15 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
}
// use local reference to avoid NPE as this.urlInvokerMap will be
set null concurrently at destroyAllInvokers().
- Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;
+ Map<ProtocolServiceKeyWithAddress, Invoker<T>> localUrlInvokerMap
= this.urlInvokerMap;
// can't use local reference as oldUrlInvokerMap's mappings might
be removed directly at toInvokers().
- Map<String, Invoker<T>> oldUrlInvokerMap = null;
+ Map<ProtocolServiceKeyWithAddress, Invoker<T>> oldUrlInvokerMap =
null;
if (localUrlInvokerMap != null) {
// the initial capacity should be set greater than the maximum
number of entries divided by the load factor to avoid resizing.
oldUrlInvokerMap = new LinkedHashMap<>(Math.round(1 +
localUrlInvokerMap.size() / DEFAULT_HASHMAP_LOAD_FACTOR));
localUrlInvokerMap.forEach(oldUrlInvokerMap::put);
}
- Map<String, Invoker<T>> newUrlInvokerMap =
toInvokers(oldUrlInvokerMap, invokerUrls);// Translate url list to Invoker map
+ Map<ProtocolServiceKeyWithAddress, Invoker<T>> newUrlInvokerMap =
toInvokers(oldUrlInvokerMap, invokerUrls);// Translate url list to Invoker map
logger.info("Refreshed invoker size " + newUrlInvokerMap.size());
if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) {
@@ -266,11 +282,12 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
* @param urls
* @return invokers
*/
- private Map<String, Invoker<T>> toInvokers(Map<String, Invoker<T>>
oldUrlInvokerMap, List<URL> urls) {
- Map<String, Invoker<T>> newUrlInvokerMap = new
ConcurrentHashMap<>(urls == null ? 1 : (int) (urls.size() / 0.75f + 1));
+ private Map<ProtocolServiceKeyWithAddress, Invoker<T>>
toInvokers(Map<ProtocolServiceKeyWithAddress, Invoker<T>> oldUrlInvokerMap,
List<URL> urls) {
+ Map<ProtocolServiceKeyWithAddress, Invoker<T>> newUrlInvokerMap = new
ConcurrentHashMap<>(urls == null ? 1 : (int) (urls.size() / 0.75f + 1));
if (urls == null || urls.isEmpty()) {
return newUrlInvokerMap;
}
+
for (URL url : urls) {
InstanceAddressURL instanceAddressURL = (InstanceAddressURL) url;
if (EMPTY_PROTOCOL.equals(instanceAddressURL.getProtocol())) {
@@ -291,33 +308,59 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
instanceAddressURL =
overrideWithConfigurator(instanceAddressURL);
}
- Invoker<T> invoker = oldUrlInvokerMap == null ? null :
oldUrlInvokerMap.get(instanceAddressURL.getAddress());
- if (invoker == null || urlChanged(invoker, instanceAddressURL)) {
// Not in the cache, refer again
- try {
- boolean enabled = true;
- if (instanceAddressURL.hasParameter(DISABLED_KEY)) {
- enabled =
!instanceAddressURL.getParameter(DISABLED_KEY, false);
- } else {
- enabled = instanceAddressURL.getParameter(ENABLED_KEY,
true);
+ // filter all the service available (version wildcard, group
wildcard, protocol wildcard)
+ int port = instanceAddressURL.getPort();
+ List<ProtocolServiceKey> matchedProtocolServiceKeys =
instanceAddressURL.getMetadataInfo()
+ .getServices().values()
+ .stream()
+ .filter(serviceInfo -> serviceInfo.getPort() <= 0 ||
serviceInfo.getPort() == port)
+ .map(MetadataInfo.ServiceInfo::getProtocolServiceKey)
+ .filter(key ->
ProtocolServiceKey.Matcher.isMatch(consumerProtocolServiceKey, key))
+ .collect(Collectors.toList());
+
+ // see org.apache.dubbo.common.ProtocolServiceKey.isSameWith
+ // check if needed to override the consumer url
+ boolean shouldWrap = matchedProtocolServiceKeys.size() != 1 ||
!consumerProtocolServiceKey.isSameWith(matchedProtocolServiceKeys.get(0));
+
+ for (ProtocolServiceKey matchedProtocolServiceKey :
matchedProtocolServiceKeys) {
+ ProtocolServiceKeyWithAddress protocolServiceKeyWithAddress =
new ProtocolServiceKeyWithAddress(matchedProtocolServiceKey,
instanceAddressURL.getAddress());
+ Invoker<T> invoker = oldUrlInvokerMap == null ? null :
oldUrlInvokerMap.get(protocolServiceKeyWithAddress);
+ if (invoker == null || urlChanged(invoker, instanceAddressURL,
matchedProtocolServiceKey)) { // Not in the cache, refer again
+ try {
+ boolean enabled;
+ if (instanceAddressURL.hasParameter(DISABLED_KEY)) {
+ enabled =
!instanceAddressURL.getParameter(DISABLED_KEY, false);
+ } else {
+ enabled =
instanceAddressURL.getParameter(ENABLED_KEY, true);
+ }
+ if (enabled) {
+ if (shouldWrap) {
+ URL newConsumerUrl =
customizedConsumerUrlMap.computeIfAbsent(matchedProtocolServiceKey,
+ k ->
consumerUrl.setProtocol(k.getProtocol())
+
.addParameter(CommonConstants.GROUP_KEY, k.getGroup())
+
.addParameter(CommonConstants.VERSION_KEY, k.getVersion()));
+
RpcContext.getServiceContext().setConsumerUrl(newConsumerUrl);
+ invoker = new
InstanceWrappedInvoker<>(protocol.refer(serviceType, instanceAddressURL),
newConsumerUrl, matchedProtocolServiceKey);
+ } else {
+ invoker = protocol.refer(serviceType,
instanceAddressURL);
+ }
+ }
+ } catch (Throwable t) {
+ logger.error("Failed to refer invoker for interface:"
+ serviceType + ",url:(" + instanceAddressURL + ")" + t.getMessage(), t);
}
- if (enabled) {
- invoker = protocol.refer(serviceType,
instanceAddressURL);
+ if (invoker != null) { // Put new invoker in cache
+ newUrlInvokerMap.put(protocolServiceKeyWithAddress,
invoker);
}
- } catch (Throwable t) {
- logger.error("Failed to refer invoker for interface:" +
serviceType + ",url:(" + instanceAddressURL + ")" + t.getMessage(), t);
+ } else {
+ newUrlInvokerMap.put(protocolServiceKeyWithAddress,
invoker);
+ oldUrlInvokerMap.remove(protocolServiceKeyWithAddress,
invoker);
}
- if (invoker != null) { // Put new invoker in cache
- newUrlInvokerMap.put(instanceAddressURL.getAddress(),
invoker);
- }
- } else {
- newUrlInvokerMap.put(instanceAddressURL.getAddress(), invoker);
- oldUrlInvokerMap.remove(instanceAddressURL.getAddress(),
invoker);
}
}
return newUrlInvokerMap;
}
- private boolean urlChanged(Invoker<T> invoker, InstanceAddressURL newURL) {
+ private boolean urlChanged(Invoker<T> invoker, InstanceAddressURL newURL,
ProtocolServiceKey protocolServiceKey) {
InstanceAddressURL oldURL = (InstanceAddressURL) invoker.getUrl();
if (!newURL.getInstance().equals(oldURL.getInstance())) {
@@ -335,16 +378,35 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
}
}
- MetadataInfo.ServiceInfo oldServiceInfo =
oldURL.getMetadataInfo().getValidServiceInfo(getConsumerUrl().getProtocolServiceKey());
+ MetadataInfo.ServiceInfo oldServiceInfo =
oldURL.getMetadataInfo().getValidServiceInfo(protocolServiceKey.toString());
if (null == oldServiceInfo) {
return false;
}
- return
!oldServiceInfo.equals(newURL.getMetadataInfo().getValidServiceInfo(getConsumerUrl().getProtocolServiceKey()));
+ return
!oldServiceInfo.equals(newURL.getMetadataInfo().getValidServiceInfo(protocolServiceKey.toString()));
}
private List<Invoker<T>> toMergeInvokerList(List<Invoker<T>> invokers) {
- return invokers;
+ List<Invoker<T>> mergedInvokers = new ArrayList<>();
+ Map<String, List<Invoker<T>>> groupMap = new HashMap<>();
+ for (Invoker<T> invoker : invokers) {
+ String group = invoker.getUrl().getGroup("");
+ groupMap.computeIfAbsent(group, k -> new ArrayList<>());
+ groupMap.get(group).add(invoker);
+ }
+
+ if (groupMap.size() == 1) {
+ mergedInvokers.addAll(groupMap.values().iterator().next());
+ } else if (groupMap.size() > 1) {
+ for (List<Invoker<T>> groupList : groupMap.values()) {
+ StaticDirectory<T> staticDirectory = new
StaticDirectory<>(groupList);
+ staticDirectory.buildRouterChain();
+ mergedInvokers.add(cluster.join(staticDirectory, false));
+ }
+ } else {
+ mergedInvokers = invokers;
+ }
+ return mergedInvokers;
}
/**
@@ -352,7 +414,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
*/
@Override
protected void destroyAllInvokers() {
- Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; //
local reference
+ Map<ProtocolServiceKeyWithAddress, Invoker<T>> localUrlInvokerMap =
this.urlInvokerMap; // local reference
if (localUrlInvokerMap != null) {
for (Invoker<T> invoker : new
ArrayList<>(localUrlInvokerMap.values())) {
try {
@@ -375,7 +437,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
* @param oldUrlInvokerMap
* @param newUrlInvokerMap
*/
- private void destroyUnusedInvokers(Map<String, Invoker<T>>
oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
+ private void destroyUnusedInvokers(Map<ProtocolServiceKeyWithAddress,
Invoker<T>> oldUrlInvokerMap, Map<ProtocolServiceKeyWithAddress, Invoker<T>>
newUrlInvokerMap) {
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
destroyAllInvokers();
return;
@@ -385,7 +447,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
return;
}
- for (Map.Entry<String, Invoker<T>> entry :
oldUrlInvokerMap.entrySet()) {
+ for (Map.Entry<ProtocolServiceKeyWithAddress, Invoker<T>> entry :
oldUrlInvokerMap.entrySet()) {
Invoker<T> invoker = entry.getValue();
if (invoker != null) {
try {
@@ -461,4 +523,88 @@ public class ServiceDiscoveryRegistryDirectory<T> extends
DynamicDirectory<T> {
});
}
}
+
+ public static final class ProtocolServiceKeyWithAddress extends
ProtocolServiceKey {
+ private final String address;
+
+ public ProtocolServiceKeyWithAddress(ProtocolServiceKey
protocolServiceKey, String address) {
+ super(protocolServiceKey.getInterfaceName(),
protocolServiceKey.getVersion(), protocolServiceKey.getGroup(),
protocolServiceKey.getProtocol());
+ this.address = address;
+ }
+
+ public String getAddress() {
+ return address;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ ProtocolServiceKeyWithAddress that =
(ProtocolServiceKeyWithAddress) o;
+ return Objects.equals(address, that.address);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), address);
+ }
+ }
+
+ public static final class InstanceWrappedInvoker<T> implements Invoker<T> {
+ private final Invoker<T> originInvoker;
+ private final URL newConsumerUrl;
+ private final ProtocolServiceKey protocolServiceKey;
+
+ public InstanceWrappedInvoker(Invoker<T> originInvoker, URL
newConsumerUrl, ProtocolServiceKey protocolServiceKey) {
+ this.originInvoker = originInvoker;
+ this.newConsumerUrl = newConsumerUrl;
+ this.protocolServiceKey = protocolServiceKey;
+ }
+
+ @Override
+ public Class<T> getInterface() {
+ return originInvoker.getInterface();
+ }
+
+ @Override
+ public Result invoke(Invocation invocation) throws RpcException {
+ // override consumer url with real protocol service key
+ RpcContext.getServiceContext().setConsumerUrl(newConsumerUrl);
+ // recreate invocation due to the protocol service key changed
+ RpcInvocation copiedInvocation = new
RpcInvocation(invocation.getTargetServiceUniqueName(),
+ invocation.getServiceModel(), invocation.getMethodName(),
invocation.getServiceName(), protocolServiceKey.toString(),
+ invocation.getParameterTypes(), invocation.getArguments(),
invocation.getObjectAttachments(),
+ invocation.getInvoker(), invocation.getAttributes(),
+ invocation instanceof RpcInvocation ? ((RpcInvocation)
invocation).getInvokeMode() : null);
+ copiedInvocation.setObjectAttachment(CommonConstants.GROUP_KEY,
protocolServiceKey.getGroup());
+ copiedInvocation.setObjectAttachment(CommonConstants.VERSION_KEY,
protocolServiceKey.getVersion());
+ return originInvoker.invoke(copiedInvocation);
+ }
+
+ @Override
+ public URL getUrl() {
+ RpcContext.getServiceContext().setConsumerUrl(newConsumerUrl);
+ return originInvoker.getUrl();
+ }
+
+ @Override
+ public boolean isAvailable() {
+ RpcContext.getServiceContext().setConsumerUrl(newConsumerUrl);
+ return originInvoker.isAvailable();
+ }
+
+ @Override
+ public void destroy() {
+ RpcContext.getServiceContext().setConsumerUrl(newConsumerUrl);
+ originInvoker.destroy();
+ }
+ }
+
}
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
index 486fd90868..0ec60af5ae 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
@@ -16,14 +16,15 @@
*/
package org.apache.dubbo.registry.client.event.listener;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder;
+import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
-import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataInfo.ServiceInfo;
import org.apache.dubbo.registry.NotifyListener;
@@ -38,7 +39,6 @@ import org.apache.dubbo.rpc.model.ScopeModelUtil;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -52,8 +52,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import static java.util.Collections.emptySet;
-import static
org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static
org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL;
import static
org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY;
@@ -77,7 +75,7 @@ public class ServiceInstancesChangedListener {
protected AtomicBoolean destroyed = new AtomicBoolean(false);
protected Map<String, List<ServiceInstance>> allInstances;
- protected Map<String, Object> serviceUrls;
+ protected Map<String, List<ProtocolServiceKeyWithUrls>> serviceUrls;
private volatile long lastRefreshTime;
private final Semaphore retryPermission;
@@ -96,8 +94,7 @@ public class ServiceInstancesChangedListener {
this.allInstances = new HashMap<>();
this.serviceUrls = new HashMap<>();
retryPermission = new Semaphore(1);
- this.scheduler = ScopeModelUtil.getApplicationModel(serviceDiscovery
== null || serviceDiscovery.getUrl() == null ? null :
serviceDiscovery.getUrl().getScopeModel())
-
.getBeanFactory().getBean(FrameworkExecutorRepository.class).getMetadataRetryExecutor();
+ this.scheduler = ScopeModelUtil.getApplicationModel(serviceDiscovery
== null || serviceDiscovery.getUrl() == null ? null :
serviceDiscovery.getUrl().getScopeModel()).getBeanFactory().getBean(FrameworkExecutorRepository.class).getMetadataRetryExecutor();
}
/**
@@ -127,7 +124,7 @@ public class ServiceInstancesChangedListener {
}
Map<String, List<ServiceInstance>> revisionToInstances = new
HashMap<>();
- Map<String, Map<String, Set<String>>> localServiceToRevisions = new
HashMap<>();
+ Map<ServiceInfo, Set<String>> localServiceToRevisions = new
HashMap<>();
// grouping all instances of this app(service name) by revision
for (Map.Entry<String, List<ServiceInstance>> entry :
allInstances.entrySet()) {
@@ -179,55 +176,45 @@ public class ServiceInstancesChangedListener {
}
hasEmptyMetadata = false;
- Map<String, Map<Set<String>, Object>> protocolRevisionsToUrls = new
HashMap<>();
- Map<String, Object> newServiceUrls = new HashMap<>();
- for (Map.Entry<String, Map<String, Set<String>>> entry :
localServiceToRevisions.entrySet()) {
- String protocol = entry.getKey();
- entry.getValue().forEach((protocolServiceKey, revisions) -> {
- Map<Set<String>, Object> revisionsToUrls =
protocolRevisionsToUrls.computeIfAbsent(protocol, k -> new HashMap<>());
- Object urls = revisionsToUrls.get(revisions);
- if (urls == null) {
- urls = getServiceUrlsCache(revisionToInstances, revisions,
protocol);
- revisionsToUrls.put(revisions, urls);
- }
+ Map<String, Map<Integer, Map<Set<String>, Object>>>
protocolRevisionsToUrls = new HashMap<>();
+ Map<String, List<ProtocolServiceKeyWithUrls>> newServiceUrls = new
HashMap<>();
+ for (Map.Entry<ServiceInfo, Set<String>> entry :
localServiceToRevisions.entrySet()) {
+ ServiceInfo serviceInfo = entry.getKey();
+ Set<String> revisions = entry.getValue();
+
+ Map<Integer, Map<Set<String>, Object>> portToRevisions =
protocolRevisionsToUrls.computeIfAbsent(serviceInfo.getProtocol(), k -> new
HashMap<>());
+ Map<Set<String>, Object> revisionsToUrls =
portToRevisions.computeIfAbsent(serviceInfo.getPort(), k -> new HashMap<>());
+ Object urls = revisionsToUrls.get(revisions);
+ if (urls == null) {
+ urls = getServiceUrlsCache(revisionToInstances, revisions,
serviceInfo.getProtocol(), serviceInfo.getPort());
+ revisionsToUrls.put(revisions, urls);
+ }
- newServiceUrls.put(protocolServiceKey, urls);
- });
+ List<ProtocolServiceKeyWithUrls> list =
newServiceUrls.computeIfAbsent(serviceInfo.getPath(), k -> new LinkedList<>());
+ list.add(new
ProtocolServiceKeyWithUrls(serviceInfo.getProtocolServiceKey(), (List<URL>)
urls));
}
this.serviceUrls = newServiceUrls;
this.notifyAddressChanged();
}
- public synchronized void addListenerAndNotify(String serviceKey,
NotifyListener listener) {
+ public synchronized void addListenerAndNotify(URL url, NotifyListener
listener) {
if (destroyed.get()) {
return;
}
- Set<NotifyListenerWithKey> notifyListeners =
this.listeners.computeIfAbsent(serviceKey, _k -> new ConcurrentHashSet<>());
- // {@code protocolServiceKeysToConsume} will be specific protocols
configured in reference config or default protocols supported by framework.
- Set<String> protocolServiceKeysToConsume =
getProtocolServiceKeyList(serviceKey, listener);
- // Add current listener to serviceKey set, there will have more than
one listener when multiple references of one same service is configured.
- NotifyListenerWithKey listenerWithKey = new
NotifyListenerWithKey(serviceKey, protocolServiceKeysToConsume, listener);
+ Set<NotifyListenerWithKey> notifyListeners =
this.listeners.computeIfAbsent(url.getServiceKey(), _k -> new
ConcurrentHashSet<>());
+ String protocol = listener.getConsumerUrl().getParameter(PROTOCOL_KEY,
url.getProtocol());
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(url.getServiceInterface(), url.getVersion(), url.getGroup(),
+ !CommonConstants.CONSUMER.equals(protocol) ? protocol : null);
+ NotifyListenerWithKey listenerWithKey = new
NotifyListenerWithKey(protocolServiceKey, listener);
notifyListeners.add(listenerWithKey);
// Aggregate address and notify on subscription.
- List<URL> urls;
- if (protocolServiceKeysToConsume.size() > 1) {
- urls = new ArrayList<>();
- for (String protocolServiceKey : protocolServiceKeysToConsume) {
- List<URL> urlsOfProtocol = getAddresses(protocolServiceKey,
listener.getConsumerUrl());
- if (CollectionUtils.isNotEmpty(urlsOfProtocol)) {
- logger.info(String.format("Found %s urls of protocol
service key %s ", urlsOfProtocol.size(), protocolServiceKey));
- urls.addAll(urlsOfProtocol);
- }
- }
- } else {
- urls =
getAddresses(protocolServiceKeysToConsume.iterator().next(),
listener.getConsumerUrl());
- }
+ List<URL> urls = getAddresses(protocolServiceKey,
listener.getConsumerUrl());
if (CollectionUtils.isNotEmpty(urls)) {
- logger.info(String.format("Notify serviceKey: %s, listener: %s
with %s urls on subscription", serviceKey, listener, urls.size()));
+ logger.info(String.format("Notify serviceKey: %s, listener: %s
with %s urls on subscription", protocolServiceKey, listener, urls.size()));
listener.notify(urls);
}
}
@@ -240,9 +227,7 @@ public class ServiceInstancesChangedListener {
// synchronized method, no need to use DCL
Set<NotifyListenerWithKey> notifyListeners =
this.listeners.get(serviceKey);
if (notifyListeners != null) {
- NotifyListenerWithKey listenerWithKey = new
NotifyListenerWithKey(serviceKey, notifyListener);
- // Remove from global listeners
- notifyListeners.remove(listenerWithKey);
+ notifyListeners.removeIf(listener ->
listener.getNotifyListener().equals(notifyListener));
// ServiceKey has no listener, remove set
if (notifyListeners.size() == 0) {
@@ -341,23 +326,28 @@ public class ServiceInstancesChangedListener {
return emptyMetadataNum;
}
- protected Map<String, Map<String, Set<String>>> parseMetadata(String
revision, MetadataInfo metadata, Map<String, Map<String, Set<String>>>
localServiceToRevisions) {
+ protected Map<ServiceInfo, Set<String>> parseMetadata(String revision,
MetadataInfo metadata, Map<ServiceInfo, Set<String>> localServiceToRevisions) {
Map<String, ServiceInfo> serviceInfos = metadata.getServices();
for (Map.Entry<String, ServiceInfo> entry : serviceInfos.entrySet()) {
- String protocol = entry.getValue().getProtocol();
- String protocolServiceKey = entry.getValue().getMatchKey();
- Map<String, Set<String>> map =
localServiceToRevisions.computeIfAbsent(protocol, _p -> new HashMap<>());
- Set<String> set = map.computeIfAbsent(protocolServiceKey, _k ->
new TreeSet<>());
+ Set<String> set =
localServiceToRevisions.computeIfAbsent(entry.getValue(), _k -> new
TreeSet<>());
set.add(revision);
}
return localServiceToRevisions;
}
- protected Object getServiceUrlsCache(Map<String, List<ServiceInstance>>
revisionToInstances, Set<String> revisions, String protocol) {
+ protected Object getServiceUrlsCache(Map<String, List<ServiceInstance>>
revisionToInstances, Set<String> revisions, String protocol, int port) {
List<URL> urls = new ArrayList<>();
for (String r : revisions) {
for (ServiceInstance i : revisionToInstances.get(r)) {
+ if (port > 0) {
+ if (i.getPort() == port) {
+
urls.add(i.toURL(protocol).setScopeModel(i.getApplicationModel()));
+ } else {
+ urls.add(((DefaultServiceInstance)
i).copyFrom(port).toURL(protocol).setScopeModel(i.getApplicationModel()));
+ }
+ continue;
+ }
// different protocols may have ports specified in meta
if (ServiceInstanceMetadataUtils.hasEndpoints(i)) {
DefaultServiceInstance.Endpoint endpoint =
ServiceInstanceMetadataUtils.getEndpoint(i, protocol);
@@ -372,8 +362,17 @@ public class ServiceInstancesChangedListener {
return urls;
}
- protected List<URL> getAddresses(String serviceProtocolKey, URL
consumerURL) {
- return (List<URL>) serviceUrls.get(serviceProtocolKey);
+ protected List<URL> getAddresses(ProtocolServiceKey protocolServiceKey,
URL consumerURL) {
+ List<ProtocolServiceKeyWithUrls> protocolServiceKeyWithUrlsList =
serviceUrls.get(protocolServiceKey.getInterfaceName());
+ List<URL> urls = new ArrayList<>();
+ if (protocolServiceKeyWithUrlsList != null) {
+ for (ProtocolServiceKeyWithUrls protocolServiceKeyWithUrls :
protocolServiceKeyWithUrlsList) {
+ if (ProtocolServiceKey.Matcher.isMatch(protocolServiceKey,
protocolServiceKeyWithUrls.getProtocolServiceKey())) {
+ urls.addAll(protocolServiceKeyWithUrls.getUrls());
+ }
+ }
+ }
+ return urls;
}
/**
@@ -385,28 +384,10 @@ public class ServiceInstancesChangedListener {
// 2 multiple subscription listener of the same service
for (NotifyListenerWithKey listenerWithKey : listenerSet) {
NotifyListener notifyListener =
listenerWithKey.getNotifyListener();
- if (listenerWithKey.getProtocolServiceKeys().size() == 1) {//
2.1 if one specific protocol is specified
- String protocolServiceKey =
listenerWithKey.getProtocolServiceKeys().iterator().next();
- //FIXME, group wildcard match
- List<URL> urls =
toUrlsWithEmpty(getAddresses(protocolServiceKey,
notifyListener.getConsumerUrl()));
- logger.info("Notify service " + protocolServiceKey + "
with urls " + urls.size());
- notifyListener.notify(urls);
- } else {// 2.2 multiple protocols or no protocol(using default
protocols) set
- List<URL> urls = new ArrayList<>();
- int effectiveProtocolNum = 0;
- for (String protocolServiceKey :
listenerWithKey.getProtocolServiceKeys()) {
- List<URL> tmpUrls = getAddresses(protocolServiceKey,
notifyListener.getConsumerUrl());
- if (CollectionUtils.isNotEmpty(tmpUrls)) {
- logger.info("Found " + urls.size() + " urls of
protocol service key " + protocolServiceKey);
- effectiveProtocolNum++;
- urls.addAll(tmpUrls);
- }
- }
- logger.info("Notify service " + serviceKey + " with " +
urls.size() + " urls from " + effectiveProtocolNum + " different protocols");
- urls = toUrlsWithEmpty(urls);
- notifyListener.notify(urls);
- }
+ List<URL> urls =
toUrlsWithEmpty(getAddresses(listenerWithKey.getProtocolServiceKey(),
notifyListener.getConsumerUrl()));
+ logger.info("Notify service " +
listenerWithKey.getProtocolServiceKey() + " with urls " + urls.size());
+ notifyListener.notify(urls);
}
});
}
@@ -421,9 +402,7 @@ public class ServiceInstancesChangedListener {
if (CollectionUtils.isEmpty(urls) && !emptyProtectionEnabled) {
// notice that the service of this.url may not be the same as
notify listener.
- URL empty = URLBuilder.from(this.url)
- .setProtocol(EMPTY_PROTOCOL)
- .build();
+ URL empty =
URLBuilder.from(this.url).setProtocol(EMPTY_PROTOCOL).build();
urls.add(empty);
}
return urls;
@@ -468,46 +447,6 @@ public class ServiceInstancesChangedListener {
return Objects.hash(getClass(), getServiceNames());
}
- /**
- * Calculate the protocol list that the consumer cares about.
- *
- * @param serviceKey possible input serviceKey includes
- * 1. {group}/{interface}:{version}, if protocol is not
specified
- * 2. {group}/{interface}:{version}:{user specified
protocols}
- * @param listener listener also contains the user specified protocols
- * @return protocol list with the format
{group}/{interface}:{version}:{protocol}
- */
- protected Set<String> getProtocolServiceKeyList(String serviceKey,
NotifyListener listener) {
- if (StringUtils.isEmpty(serviceKey)) {
- return emptySet();
- }
-
- Set<String> result = new HashSet<>();
- String protocol = listener.getConsumerUrl().getParameter(PROTOCOL_KEY);
- if (serviceKey.endsWith(CONSUMER_PROTOCOL_SUFFIX)) {
- serviceKey = serviceKey.substring(0,
serviceKey.indexOf(CONSUMER_PROTOCOL_SUFFIX));
- }
-
- if (StringUtils.isNotEmpty(protocol)) {
- int protocolIndex = serviceKey.indexOf(":" + protocol);
- if (protocol.contains(",") && protocolIndex != -1) {
- serviceKey = serviceKey.substring(0, protocolIndex);
- String[] specifiedProtocols = protocol.split(",");
- for (String specifiedProtocol : specifiedProtocols) {
- result.add(serviceKey + GROUP_CHAR_SEPARATOR +
specifiedProtocol);
- }
- } else {
- result.add(serviceKey);
- }
- } else {
- for (String supportedProtocol : SUPPORTED_PROTOCOLS) {
- result.add(serviceKey + GROUP_CHAR_SEPARATOR +
supportedProtocol);
- }
- }
-
- return result;
- }
-
protected class AddressRefreshRetryTask implements Runnable {
private final RetryServiceInstancesChangedEvent retryEvent;
private final Semaphore retryPermission;
@@ -525,26 +464,16 @@ public class ServiceInstancesChangedListener {
}
public static class NotifyListenerWithKey {
- private final String serviceKey;
- private final Set<String> protocolServiceKeys;
+ private final ProtocolServiceKey protocolServiceKey;
private final NotifyListener notifyListener;
- public NotifyListenerWithKey(String protocolServiceKey, Set<String>
protocolServiceKeys, NotifyListener notifyListener) {
- this.serviceKey = protocolServiceKey;
- this.protocolServiceKeys = (protocolServiceKeys == null ? new
ConcurrentHashSet<>() : protocolServiceKeys);
+ public NotifyListenerWithKey(ProtocolServiceKey protocolServiceKey,
NotifyListener notifyListener) {
+ this.protocolServiceKey = protocolServiceKey;
this.notifyListener = notifyListener;
}
- public NotifyListenerWithKey(String protocolServiceKey, NotifyListener
notifyListener) {
- this(protocolServiceKey, null, notifyListener);
- }
-
- public String getServiceKey() {
- return serviceKey;
- }
-
- public Set<String> getProtocolServiceKeys() {
- return protocolServiceKeys;
+ public ProtocolServiceKey getProtocolServiceKey() {
+ return protocolServiceKey;
}
public NotifyListener getNotifyListener() {
@@ -560,12 +489,30 @@ public class ServiceInstancesChangedListener {
return false;
}
NotifyListenerWithKey that = (NotifyListenerWithKey) o;
- return Objects.equals(serviceKey, that.serviceKey) &&
Objects.equals(notifyListener, that.notifyListener);
+ return Objects.equals(protocolServiceKey, that.protocolServiceKey)
&& Objects.equals(notifyListener, that.notifyListener);
}
@Override
public int hashCode() {
- return Objects.hash(serviceKey, notifyListener);
+ return Objects.hash(protocolServiceKey, notifyListener);
+ }
+ }
+
+ public static class ProtocolServiceKeyWithUrls {
+ private final ProtocolServiceKey protocolServiceKey;
+ private final List<URL> urls;
+
+ public ProtocolServiceKeyWithUrls(ProtocolServiceKey
protocolServiceKey, List<URL> urls) {
+ this.protocolServiceKey = protocolServiceKey;
+ this.urls = urls;
+ }
+
+ public ProtocolServiceKey getProtocolServiceKey() {
+ return protocolServiceKey;
+ }
+
+ public List<URL> getUrls() {
+ return urls;
}
}
}
diff --git
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryTest.java
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryTest.java
index 6ce0410b43..40546c73c2 100644
---
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryTest.java
+++
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.registry.client;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.metadata.AbstractServiceNameMapping;
import org.apache.dubbo.metadata.ServiceNameMapping;
@@ -43,7 +44,6 @@ import java.util.concurrent.locks.ReentrantLock;
import static org.apache.dubbo.common.constants.CommonConstants.CHECK_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
-import static
org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static org.apache.dubbo.metadata.ServiceNameMapping.toStringKeys;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -220,10 +220,10 @@ public class ServiceDiscoveryRegistryTest {
// check different protocol
Map<String,
Set<ServiceInstancesChangedListener.NotifyListenerWithKey>> serviceListeners =
multiAppsInstanceListener.getServiceListeners();
assertEquals(2, serviceListeners.size());
- assertEquals(1,
serviceListeners.get(url.getProtocolServiceKey()).size());
- assertEquals(1,
serviceListeners.get(url2.getProtocolServiceKey()).size());
- String protocolServiceKey = url2.getServiceKey() +
GROUP_CHAR_SEPARATOR + url2.getParameter(PROTOCOL_KEY, DUBBO);
-
assertTrue(serviceListeners.get(url2.getProtocolServiceKey()).contains(new
ServiceInstancesChangedListener.NotifyListenerWithKey(protocolServiceKey,
testServiceListener2)));
+ assertEquals(1, serviceListeners.get(url.getServiceKey()).size());
+ assertEquals(1, serviceListeners.get(url2.getServiceKey()).size());
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(url2.getServiceInterface(), url2.getVersion(),
url2.getGroup(), url2.getParameter(PROTOCOL_KEY, DUBBO));
+ assertTrue(serviceListeners.get(url2.getServiceKey()).contains(new
ServiceInstancesChangedListener.NotifyListenerWithKey(protocolServiceKey,
testServiceListener2)));
}
/**
diff --git
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/MockServiceInstancesChangedListener.java
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/MockServiceInstancesChangedListener.java
index 7c187eeef1..26f82b4f5b 100644
---
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/MockServiceInstancesChangedListener.java
+++
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/MockServiceInstancesChangedListener.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.registry.client.event.listener;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
@@ -35,8 +36,8 @@ public class MockServiceInstancesChangedListener extends
ServiceInstancesChanged
}
@Override
- public List<URL> getAddresses(String serviceProtocolKey, URL consumerURL) {
- return super.getAddresses(serviceProtocolKey, consumerURL);
+ public List<URL> getAddresses(ProtocolServiceKey protocolServiceKey, URL
consumerURL) {
+ return super.getAddresses(protocolServiceKey, consumerURL);
}
public Map<String, Set<NotifyListenerWithKey>> getServiceListeners() {
diff --git
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java
index b8c69614c1..7bab0b8554 100644
---
a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java
+++
b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListenerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.dubbo.registry.client.event.listener;
+import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.common.utils.LRUCache;
@@ -49,7 +50,6 @@ import org.mockito.stubbing.Answer;
import java.lang.reflect.Field;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -62,7 +62,6 @@ import static
org.apache.dubbo.common.utils.CollectionUtils.isEmpty;
import static org.apache.dubbo.common.utils.CollectionUtils.isNotEmpty;
import static
org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.EXPORTED_SERVICES_REVISION_PROPERTY_NAME;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.eq;
@@ -219,7 +218,8 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(1, allInstances.size());
Assertions.assertEquals(3, allInstances.get("app1").size());
- List<URL> serviceUrls = listener.getAddresses(service1 + ":dubbo",
consumerURL);
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey,
consumerURL);
Assertions.assertEquals(3, serviceUrls.size());
assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
}
@@ -240,7 +240,8 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(1, allInstances.size());
Assertions.assertEquals(3, allInstances.get("app1").size());
- List<URL> serviceUrls = listener.getAddresses(service1 + ":dubbo",
consumerURL);
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey,
consumerURL);
Assertions.assertEquals(3, serviceUrls.size());
assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
@@ -271,12 +272,16 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(3, allInstances.get("app1").size());
Assertions.assertEquals(4, allInstances.get("app2").size());
- List<URL> serviceUrls = listener.getAddresses(service1 + ":dubbo",
consumerURL);
+ ProtocolServiceKey protocolServiceKey1 = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey2 = new
ProtocolServiceKey(service2, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey3 = new
ProtocolServiceKey(service3, null, null, "dubbo");
+
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey1,
consumerURL);
Assertions.assertEquals(7, serviceUrls.size());
- List<URL> serviceUrls2 = listener.getAddresses(service2 + ":dubbo",
consumerURL);
+ List<URL> serviceUrls2 = listener.getAddresses(protocolServiceKey2,
consumerURL);
Assertions.assertEquals(4, serviceUrls2.size());
assertTrue(serviceUrls2.get(0).getIp().contains("30.10."));
- List<URL> serviceUrls3 = listener.getAddresses(service3 + ":dubbo",
consumerURL);
+ List<URL> serviceUrls3 = listener.getAddresses(protocolServiceKey3,
consumerURL);
Assertions.assertEquals(2, serviceUrls3.size());
assertTrue(serviceUrls3.get(0).getIp().contains("30.10."));
}
@@ -308,13 +313,17 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(0, allInstances.get("app1").size());
Assertions.assertEquals(4, allInstances.get("app2").size());
- List<URL> serviceUrls = listener.getAddresses(service1 + ":dubbo",
consumerURL);
+ ProtocolServiceKey protocolServiceKey1 = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey2 = new
ProtocolServiceKey(service2, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey3 = new
ProtocolServiceKey(service3, null, null, "dubbo");
+
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey1,
consumerURL);
Assertions.assertEquals(4, serviceUrls.size());
assertTrue(serviceUrls.get(0).getIp().contains("30.10."));
- List<URL> serviceUrls2 = listener.getAddresses(service2 + ":dubbo",
consumerURL);
+ List<URL> serviceUrls2 = listener.getAddresses(protocolServiceKey2,
consumerURL);
Assertions.assertEquals(4, serviceUrls2.size());
assertTrue(serviceUrls2.get(0).getIp().contains("30.10."));
- List<URL> serviceUrls3 = listener.getAddresses(service3 + ":dubbo",
consumerURL);
+ List<URL> serviceUrls3 = listener.getAddresses(protocolServiceKey3,
consumerURL);
Assertions.assertEquals(2, serviceUrls3.size());
assertTrue(serviceUrls3.get(0).getIp().contains("30.10."));
@@ -328,9 +337,9 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(0, allInstances_app2.get("app1").size());
Assertions.assertEquals(0, allInstances_app2.get("app2").size());
- assertTrue(isEmpty(listener.getAddresses(service1 + ":dubbo",
consumerURL)));
- assertTrue(isEmpty(listener.getAddresses(service2 + ":dubbo",
consumerURL)));
- assertTrue(isEmpty(listener.getAddresses(service3 + ":dubbo",
consumerURL)));
+ assertTrue(isEmpty(listener.getAddresses(protocolServiceKey1,
consumerURL)));
+ assertTrue(isEmpty(listener.getAddresses(protocolServiceKey2,
consumerURL)));
+ assertTrue(isEmpty(listener.getAddresses(protocolServiceKey3,
consumerURL)));
}
// 正常场景。检查instance listener -> service listener(Directory)地址推送流程
@@ -345,8 +354,8 @@ public class ServiceInstancesChangedListenerTest {
when(demoServiceListener.getConsumerUrl()).thenReturn(consumerURL);
NotifyListener demoService2Listener =
Mockito.mock(NotifyListener.class);
when(demoService2Listener.getConsumerUrl()).thenReturn(consumerURL2);
- listener.addListenerAndNotify(consumerURL.getProtocolServiceKey(),
demoServiceListener);
- listener.addListenerAndNotify(consumerURL2.getProtocolServiceKey(),
demoService2Listener);
+ listener.addListenerAndNotify(consumerURL, demoServiceListener);
+ listener.addListenerAndNotify(consumerURL2, demoService2Listener);
// notify app1 instance change
ServiceInstancesChangedEvent app1_event = new
ServiceInstancesChangedEvent("app1", app1Instances);
listener.onEvent(app1_event);
@@ -378,7 +387,7 @@ public class ServiceInstancesChangedListenerTest {
// test service listener still get notified when added after instance
notification.
NotifyListener demoService3Listener =
Mockito.mock(NotifyListener.class);
when(demoService3Listener.getConsumerUrl()).thenReturn(consumerURL3);
- listener.addListenerAndNotify(consumerURL3.getProtocolServiceKey(),
demoService3Listener);
+ listener.addListenerAndNotify(consumerURL3, demoService3Listener);
Mockito.verify(demoService3Listener,
Mockito.times(1)).notify(Mockito.anyList());
}
@@ -397,10 +406,10 @@ public class ServiceInstancesChangedListenerTest {
when(demoService2Listener1.getConsumerUrl()).thenReturn(consumerURL2);
NotifyListener demoService2Listener2 =
Mockito.mock(NotifyListener.class);
when(demoService2Listener2.getConsumerUrl()).thenReturn(consumerURL2);
- listener.addListenerAndNotify(consumerURL.getProtocolServiceKey(),
demoServiceListener1);
- listener.addListenerAndNotify(consumerURL.getProtocolServiceKey(),
demoServiceListener2);
- listener.addListenerAndNotify(consumerURL2.getProtocolServiceKey(),
demoService2Listener1);
- listener.addListenerAndNotify(consumerURL2.getProtocolServiceKey(),
demoService2Listener2);
+ listener.addListenerAndNotify(consumerURL, demoServiceListener1);
+ listener.addListenerAndNotify(consumerURL, demoServiceListener2);
+ listener.addListenerAndNotify(consumerURL2, demoService2Listener1);
+ listener.addListenerAndNotify(consumerURL2, demoService2Listener2);
// notify app1 instance change
ServiceInstancesChangedEvent app1_event = new
ServiceInstancesChangedEvent("app1", app1Instances);
listener.onEvent(app1_event);
@@ -432,7 +441,7 @@ public class ServiceInstancesChangedListenerTest {
// test service listener still get notified when added after instance
notification.
NotifyListener demoService3Listener =
Mockito.mock(NotifyListener.class);
when(demoService3Listener.getConsumerUrl()).thenReturn(consumerURL3);
- listener.addListenerAndNotify(consumerURL3.getProtocolServiceKey(),
demoService3Listener);
+ listener.addListenerAndNotify(consumerURL3, demoService3Listener);
Mockito.verify(demoService3Listener,
Mockito.times(1)).notify(Mockito.anyList());
}
@@ -448,15 +457,15 @@ public class ServiceInstancesChangedListenerTest {
// no protocol specified, consume all instances
NotifyListener demoServiceListener1 =
Mockito.mock(NotifyListener.class);
when(demoServiceListener1.getConsumerUrl()).thenReturn(noProtocolConsumerURL);
-
listener.addListenerAndNotify(noProtocolConsumerURL.getProtocolServiceKey(),
demoServiceListener1);
+ listener.addListenerAndNotify(noProtocolConsumerURL,
demoServiceListener1);
// multiple protocols specified
NotifyListener demoServiceListener2 =
Mockito.mock(NotifyListener.class);
when(demoServiceListener2.getConsumerUrl()).thenReturn(multipleProtocolsConsumerURL);
-
listener.addListenerAndNotify(multipleProtocolsConsumerURL.getProtocolServiceKey(),
demoServiceListener2);
+ listener.addListenerAndNotify(multipleProtocolsConsumerURL,
demoServiceListener2);
// one protocol specified
NotifyListener demoServiceListener3 =
Mockito.mock(NotifyListener.class);
when(demoServiceListener3.getConsumerUrl()).thenReturn(singleProtocolsConsumerURL);
-
listener.addListenerAndNotify(singleProtocolsConsumerURL.getProtocolServiceKey(),
demoServiceListener3);
+ listener.addListenerAndNotify(singleProtocolsConsumerURL,
demoServiceListener3);
// notify app1 instance change
ServiceInstancesChangedEvent app1_event = new
ServiceInstancesChangedEvent("app1", app1InstancesMultipleProtocols);
@@ -479,9 +488,119 @@ public class ServiceInstancesChangedListenerTest {
Assertions.assertEquals(1, single_protocol_notifiedUrls.size());
}
- // revision 异常场景。第一次启动，完全拿不到metadata，只能通知部分地址
+ /**
+ * Test subscribe multiple groups
+ */
@Test
@Order(8)
+ public void testSubscribeMultipleGroups() {
+ Set<String> serviceNames = new HashSet<>();
+ serviceNames.add("app1");
+ listener = new ServiceInstancesChangedListener(serviceNames,
serviceDiscovery);
+
+ // notify instance change
+ ServiceInstancesChangedEvent event = new
ServiceInstancesChangedEvent("app1", app1Instances);
+ listener.onEvent(event);
+
+ Map<String, List<ServiceInstance>> allInstances =
listener.getAllInstances();
+ Assertions.assertEquals(1, allInstances.size());
+ Assertions.assertEquals(3, allInstances.get("app1").size());
+
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey,
consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null, "",
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null, ",group1",
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null, "group1,",
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null, "*",
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null, "group1",
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(0, serviceUrls.size());
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null,
"group1,group2", "dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(0, serviceUrls.size());
+
+ protocolServiceKey = new ProtocolServiceKey(service1, null,
"group1,,group2", "dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+ }
+
+ /**
+ * Test subscribe multiple versions
+ */
+ @Test
+ @Order(9)
+ public void testSubscribeMultipleVersions() {
+ Set<String> serviceNames = new HashSet<>();
+ serviceNames.add("app1");
+ listener = new ServiceInstancesChangedListener(serviceNames,
serviceDiscovery);
+
+ // notify instance change
+ ServiceInstancesChangedEvent event = new
ServiceInstancesChangedEvent("app1", app1Instances);
+ listener.onEvent(event);
+
+ Map<String, List<ServiceInstance>> allInstances =
listener.getAllInstances();
+ Assertions.assertEquals(1, allInstances.size());
+ Assertions.assertEquals(3, allInstances.get("app1").size());
+
+ ProtocolServiceKey protocolServiceKey = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey,
consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, "", null,
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, "*", null,
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, ",1.0.0", null,
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, "1.0.0,", null,
"dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, "1.0.0,,1.0.1",
null, "dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(3, serviceUrls.size());
+ assertTrue(serviceUrls.get(0) instanceof InstanceAddressURL);
+
+ protocolServiceKey = new ProtocolServiceKey(service1, "1.0.1,1.0.0",
null, "dubbo");
+ serviceUrls = listener.getAddresses(protocolServiceKey, consumerURL);
+ Assertions.assertEquals(0, serviceUrls.size());
+ }
+
+ // revision 异常场景。第一次启动，完全拿不到metadata，只能通知部分地址
+ @Test
+ @Order(10)
public void testRevisionFailureOnStartup() {
Set<String> serviceNames = new HashSet<>();
serviceNames.add("app1");
@@ -490,8 +609,12 @@ public class ServiceInstancesChangedListenerTest {
ServiceInstancesChangedEvent failed_revision_event = new
ServiceInstancesChangedEvent("app1", app1FailedInstances);
listener.onEvent(failed_revision_event);
- List<URL> serviceUrls = listener.getAddresses(service1 + ":dubbo",
consumerURL);
- List<URL> serviceUrls2 = listener.getAddresses(service2 + ":dubbo",
consumerURL);
+
+ ProtocolServiceKey protocolServiceKey1 = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey2 = new
ProtocolServiceKey(service2, null, null, "dubbo");
+
+ List<URL> serviceUrls = listener.getAddresses(protocolServiceKey1,
consumerURL);
+ List<URL> serviceUrls2 = listener.getAddresses(protocolServiceKey2,
consumerURL);
assertTrue(isNotEmpty(serviceUrls));
assertTrue(isNotEmpty(serviceUrls2));
@@ -499,7 +622,7 @@ public class ServiceInstancesChangedListenerTest {
// revision 异常场景。运行中地址通知，拿不到revision就用老版本revision
@Test
- @Order(9)
+ @Order(11)
public void testRevisionFailureOnNotification() {
Set<String> serviceNames = new HashSet<>();
serviceNames.add("app1");
@@ -524,8 +647,11 @@ public class ServiceInstancesChangedListenerTest {
listener.onEvent(event2);
// event2 did not really take effect
- Assertions.assertEquals(3, listener.getAddresses(service1 + ":dubbo",
consumerURL).size());
- assertTrue(isEmpty(listener.getAddresses(service2 + ":dubbo",
consumerURL)));
+ ProtocolServiceKey protocolServiceKey1 = new
ProtocolServiceKey(service1, null, null, "dubbo");
+ ProtocolServiceKey protocolServiceKey2 = new
ProtocolServiceKey(service2, null, null, "dubbo");
+
+ Assertions.assertEquals(3, listener.getAddresses(protocolServiceKey1,
consumerURL).size());
+ assertTrue(isEmpty(listener.getAddresses(protocolServiceKey2,
consumerURL)));
//
init();
@@ -536,16 +662,16 @@ public class ServiceInstancesChangedListenerTest {
e.printStackTrace();
}
// check recovered after retry.
- List<URL> serviceUrls_after_retry = listener.getAddresses(service1 +
":dubbo", consumerURL);
+ List<URL> serviceUrls_after_retry =
listener.getAddresses(protocolServiceKey1, consumerURL);
Assertions.assertEquals(5, serviceUrls_after_retry.size());
- List<URL> serviceUrls2_after_retry = listener.getAddresses(service2 +
":dubbo", consumerURL);
+ List<URL> serviceUrls2_after_retry =
listener.getAddresses(protocolServiceKey2, consumerURL);
Assertions.assertEquals(2, serviceUrls2_after_retry.size());
}
// Abnormal case. Instance does not have revision
@Test
- @Order(10)
+ @Order(12)
public void testInstanceWithoutRevision() {
Set<String> serviceNames = new HashSet<>();
serviceNames.add("app1");
@@ -559,39 +685,6 @@ public class ServiceInstancesChangedListenerTest {
assertTrue(true);
}
- /**
- * Test calculation of subscription protocols
- */
- @Test
- public void testGetProtocolServiceKeyList() {
- NotifyListener listener = Mockito.mock(NotifyListener.class);
-
- Set<String> serviceNames = new HashSet<>();
- serviceNames.add("app1");
- ServiceDiscovery serviceDiscovery =
Mockito.mock(ServiceDiscovery.class);
- ServiceInstancesChangedListener instancesChangedListener = new
ServiceInstancesChangedListener(serviceNames, serviceDiscovery);
-
- URL url1 = URL.valueOf("tri://localhost/Service?protocol=tri");
- when(listener.getConsumerUrl()).thenReturn(url1);
- Set<String> keyList11 =
instancesChangedListener.getProtocolServiceKeyList(url1.getProtocolServiceKey(),
listener);
- assertEquals(getExpectedSet(Arrays.asList("Service:tri")), keyList11);
-
- URL url2 =
URL.valueOf("consumer://localhost/Service?group=group&version=1.0");
- when(listener.getConsumerUrl()).thenReturn(url2);
- Set<String> keyList12 =
instancesChangedListener.getProtocolServiceKeyList(url2.getProtocolServiceKey(),
listener);
- assertEquals(getExpectedSet(Arrays.asList("group/Service:1.0:tri",
"group/Service:1.0:dubbo", "group/Service:1.0:rest")), keyList12);
-
- URL url3 =
URL.valueOf("dubbo://localhost/Service?protocol=dubbo&group=group&version=1.0");
- when(listener.getConsumerUrl()).thenReturn(url3);
- Set<String> keyList21 =
instancesChangedListener.getProtocolServiceKeyList(url3.getProtocolServiceKey(),
listener);
- assertEquals(getExpectedSet(Arrays.asList("group/Service:1.0:dubbo")),
keyList21);
-
- URL url4 =
URL.valueOf("dubbo,tri://localhost/Service?protocol=dubbo,tri&group=group&version=1.0");
- when(listener.getConsumerUrl()).thenReturn(url4);
- Set<String> keyList23 =
instancesChangedListener.getProtocolServiceKeyList(url4.getProtocolServiceKey(),
listener);
- assertEquals(getExpectedSet(Arrays.asList("group/Service:1.0:dubbo",
"group/Service:1.0:tri")), keyList23);
- }
-
Set<String> getExpectedSet(List<String> list) {
return new HashSet<>(list);
}
