alibaba · KomachiSion · Dec 26, 2020 · Dec 26, 2020
diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/InstanceOperatorServiceImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/core/InstanceOperatorServiceImpl.java
@@ -26,10 +26,10 @@
 import com.alibaba.nacos.naming.misc.Loggers;
 import com.alibaba.nacos.naming.misc.SwitchDomain;
 import com.alibaba.nacos.naming.pojo.Subscriber;
-import com.alibaba.nacos.naming.push.ClientInfo;
-import com.alibaba.nacos.naming.push.DataSource;
-import com.alibaba.nacos.naming.push.NamingSubscriberServiceV1Impl;
-import com.alibaba.nacos.naming.push.PushClient;
+import com.alibaba.nacos.naming.push.v1.ClientInfo;
+import com.alibaba.nacos.naming.push.v1.DataSource;
+import com.alibaba.nacos.naming.push.v1.NamingSubscriberServiceV1Impl;
+import com.alibaba.nacos.naming.push.v1.PushClient;
 import com.alibaba.nacos.naming.push.UdpPushService;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;

diff --git a/naming/src/main/java/com/alibaba/nacos/naming/core/SubscribeManager.java b/naming/src/main/java/com/alibaba/nacos/naming/core/SubscribeManager.java
@@ -16,26 +16,15 @@
 
 package com.alibaba.nacos.naming.core;
 
-import com.alibaba.nacos.api.naming.CommonParams;
-import com.alibaba.nacos.common.model.RestResult;
-import com.alibaba.nacos.common.utils.JacksonUtils;
-import com.alibaba.nacos.core.cluster.Member;
-import com.alibaba.nacos.core.cluster.ServerMemberManager;
-import com.alibaba.nacos.naming.push.NamingSubscriberServiceV1Impl;
-import com.alibaba.nacos.sys.env.EnvUtil;
-import com.alibaba.nacos.naming.misc.HttpClient;
-import com.alibaba.nacos.naming.misc.NetUtils;
-import com.alibaba.nacos.naming.misc.UtilsAndCommons;
 import com.alibaba.nacos.naming.pojo.Subscriber;
-import com.alibaba.nacos.naming.pojo.Subscribers;
-import com.alibaba.nacos.naming.push.v2.NamingSubscriberServiceV2Impl;
+import com.alibaba.nacos.naming.push.NamingSubscriberServiceAggregationImpl;
+import com.alibaba.nacos.naming.push.NamingSubscriberServiceLocalImpl;
 import org.apache.commons.collections.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -48,35 +37,17 @@
  * Subscribe manager.
  *
  * @author Nicholas
+ * @author xiweng.yy
  * @since 1.0.1
  */
 @Service
 public class SubscribeManager {
 
-    private static final String SUBSCRIBER_ON_SYNC_URL = "/service/subscribers";
-
-    @Autowired
-    private NamingSubscriberServiceV1Impl subscriberServiceV1;
-
     @Autowired
-    private NamingSubscriberServiceV2Impl subscriberServiceV2;
+    private NamingSubscriberServiceLocalImpl localService;
 
     @Autowired
-    private ServerMemberManager memberManager;
-
-    private List<Subscriber> getSubscribersFuzzy(String serviceName, String namespaceId) {
-        List<Subscriber> result = new LinkedList<>();
-        result.addAll(subscriberServiceV1.getFuzzySubscribers(namespaceId, serviceName));
-        result.addAll(subscriberServiceV2.getFuzzySubscribers(namespaceId, serviceName));
-        return result;
-    }
-
-    private List<Subscriber> getSubscribers(String serviceName, String namespaceId) {
-        List<Subscriber> result = new LinkedList<>();
-        result.addAll(subscriberServiceV1.getSubscribers(namespaceId, serviceName));
-        result.addAll(subscriberServiceV2.getSubscribers(namespaceId, serviceName));
-        return result;
-    }
+    private NamingSubscriberServiceAggregationImpl aggregationService;
 
     /**
      * Get subscribers.
@@ -85,44 +56,14 @@ private List<Subscriber> getSubscribers(String serviceName, String namespaceId)
      * @param namespaceId namespace id
      * @param aggregation aggregation
      * @return list of subscriber
-     * @throws InterruptedException interrupted exception
      */
-    public List<Subscriber> getSubscribers(String serviceName, String namespaceId, boolean aggregation)
-            throws InterruptedException {
+    public List<Subscriber> getSubscribers(String serviceName, String namespaceId, boolean aggregation) {
         if (aggregation) {
-            // size = 1 means only myself in the list, we need at least one another server alive:
-            if (memberManager.getServerList().size() <= 1) {
-                return getSubscribersFuzzy(serviceName, namespaceId);
-            }
-
-            List<Subscriber> subscriberList = new ArrayList<Subscriber>();
-            // try sync data from remote server:
-            for (Member server : memberManager.allMembers()) {
-
-                Map<String, String> paramValues = new HashMap<>(128);
-                paramValues.put(CommonParams.SERVICE_NAME, serviceName);
-                paramValues.put(CommonParams.NAMESPACE_ID, namespaceId);
-                paramValues.put("aggregation", String.valueOf(Boolean.FALSE));
-                if (NetUtils.localServer().equals(server.getAddress())) {
-                    subscriberList.addAll(getSubscribersFuzzy(serviceName, namespaceId));
-                    continue;
-                }
-
-                RestResult<String> result = HttpClient.httpGet(
-                        "http://" + server.getAddress() + EnvUtil.getContextPath()
-                                + UtilsAndCommons.NACOS_NAMING_CONTEXT + SUBSCRIBER_ON_SYNC_URL, new ArrayList<>(),
-                        paramValues);
-
-                if (result.ok()) {
-                    Subscribers subscribers = JacksonUtils.toObj(result.getData(), Subscribers.class);
-                    subscriberList.addAll(subscribers.getSubscribers());
-                }
-            }
-            return CollectionUtils.isNotEmpty(subscriberList) ? subscriberList.stream()
-                    .filter(distinctByKey(Subscriber::toString)).collect(Collectors.toList()) : Collections.EMPTY_LIST;
+            Collection<Subscriber> result = aggregationService.getFuzzySubscribers(namespaceId, serviceName);
+            return CollectionUtils.isNotEmpty(result) ? result.stream().filter(distinctByKey(Subscriber::toString))
+                    .collect(Collectors.toList()) : Collections.EMPTY_LIST;
         } else {
-            // local server
-            return getSubscribersFuzzy(serviceName, namespaceId);
+            return new LinkedList<>(localService.getFuzzySubscribers(namespaceId, serviceName));
         }
     }
 

diff --git a/...g/src/main/java/com/alibaba/nacos/naming/push/NamingSubscriberServiceAggregationImpl.java b/...g/src/main/java/com/alibaba/nacos/naming/push/NamingSubscriberServiceAggregationImpl.java
@@ -16,11 +16,23 @@
 
 package com.alibaba.nacos.naming.push;
 
+import com.alibaba.nacos.api.naming.CommonParams;
+import com.alibaba.nacos.common.model.RestResult;
+import com.alibaba.nacos.common.utils.JacksonUtils;
+import com.alibaba.nacos.core.cluster.Member;
+import com.alibaba.nacos.core.cluster.ServerMemberManager;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
+import com.alibaba.nacos.naming.misc.HttpClient;
+import com.alibaba.nacos.naming.misc.UtilsAndCommons;
 import com.alibaba.nacos.naming.pojo.Subscriber;
-import com.alibaba.nacos.naming.push.v2.NamingSubscriberServiceV2Impl;
+import com.alibaba.nacos.naming.pojo.Subscribers;
+import com.alibaba.nacos.sys.env.EnvUtil;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
 
 /**
  * Aggregation naming subscriber service. Aggregate all implementation of {@link NamingSubscriberService} and
@@ -31,33 +43,70 @@
 @org.springframework.stereotype.Service
 public class NamingSubscriberServiceAggregationImpl implements NamingSubscriberService {
 
-    private final NamingSubscriberServiceV1Impl subscriberServiceV1;
+    private static final String SUBSCRIBER_ON_SYNC_URL = "/service/subscribers";
 
-    private final NamingSubscriberServiceV2Impl subscriberServiceV2;
+    private final NamingSubscriberServiceLocalImpl subscriberServiceLocal;
 
-    public NamingSubscriberServiceAggregationImpl(NamingSubscriberServiceV1Impl subscriberServiceV1,
-            NamingSubscriberServiceV2Impl subscriberServiceV2) {
-        this.subscriberServiceV1 = subscriberServiceV1;
-        this.subscriberServiceV2 = subscriberServiceV2;
+    private final ServerMemberManager memberManager;
+
+    public NamingSubscriberServiceAggregationImpl(NamingSubscriberServiceLocalImpl subscriberServiceLocal,
+            ServerMemberManager serverMemberManager) {
+        this.subscriberServiceLocal = subscriberServiceLocal;
+        this.memberManager = serverMemberManager;
     }
 
     @Override
     public Collection<Subscriber> getSubscribers(String namespaceId, String serviceName) {
-        return null;
+        Collection<Subscriber> result = new LinkedList<>(
+                subscriberServiceLocal.getSubscribers(namespaceId, serviceName));
+        if (memberManager.getServerList().size() > 1) {
+            getSubscribersFromRemotes(namespaceId, serviceName, result);
+        }
+        return result;
     }
 
     @Override
     public Collection<Subscriber> getSubscribers(Service service) {
-        return null;
+        Collection<Subscriber> result = new LinkedList<>(subscriberServiceLocal.getSubscribers(service));
+        if (memberManager.getServerList().size() > 1) {
+            getSubscribersFromRemotes(service.getNamespace(), service.getGroupedServiceName(), result);
+        }
+        return result;
     }
 
     @Override
     public Collection<Subscriber> getFuzzySubscribers(String namespaceId, String serviceName) {
-        return null;
+        Collection<Subscriber> result = new LinkedList<>(
+                subscriberServiceLocal.getFuzzySubscribers(namespaceId, serviceName));
+        if (memberManager.getServerList().size() > 1) {
+            getSubscribersFromRemotes(namespaceId, serviceName, result);
+        }
+        return result;
     }
 
     @Override
     public Collection<Subscriber> getFuzzySubscribers(Service service) {
-        return null;
+        Collection<Subscriber> result = new LinkedList<>(subscriberServiceLocal.getFuzzySubscribers(service));
+        if (memberManager.getServerList().size() > 1) {
+            getSubscribersFromRemotes(service.getNamespace(), service.getGroupedServiceName(), result);
+        }
+        return result;
+    }
+
+    private void getSubscribersFromRemotes(String namespaceId, String serviceName, Collection<Subscriber> result) {
+        for (Member server : memberManager.allMembersWithoutSelf()) {
+            Map<String, String> paramValues = new HashMap<>(128);
+            paramValues.put(CommonParams.SERVICE_NAME, serviceName);
+            paramValues.put(CommonParams.NAMESPACE_ID, namespaceId);
+            paramValues.put("aggregation", String.valueOf(Boolean.FALSE));
+            // TODO replace with gRPC
+            RestResult<String> response = HttpClient.httpGet(
+                    "http://" + server.getAddress() + EnvUtil.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT
+                            + SUBSCRIBER_ON_SYNC_URL, new ArrayList<>(), paramValues);
+            if (response.ok()) {
+                Subscribers subscribers = JacksonUtils.toObj(response.getData(), Subscribers.class);
+                result.addAll(subscribers.getSubscribers());
+            }
+        }
     }
 }
diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/NamingSubscriberServiceLocalImpl.java b/naming/src/main/java/com/alibaba/nacos/naming/push/NamingSubscriberServiceLocalImpl.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.nacos.naming.push;
+
+import com.alibaba.nacos.naming.core.v2.pojo.Service;
+import com.alibaba.nacos.naming.pojo.Subscriber;
+import com.alibaba.nacos.naming.push.v1.NamingSubscriberServiceV1Impl;
+import com.alibaba.nacos.naming.push.v2.NamingSubscriberServiceV2Impl;
+
+import java.util.Collection;
+import java.util.HashSet;
+
+/**
+ * Naming subscriber service for local.
+ *
+ * @author xiweng.yy
+ * @deprecated Will be removed with {@link com.alibaba.nacos.naming.push.v1.NamingSubscriberServiceV1Impl}
+ */
+@org.springframework.stereotype.Service
+@Deprecated
+public class NamingSubscriberServiceLocalImpl implements NamingSubscriberService {
+
+    private final NamingSubscriberServiceV1Impl namingSubscriberServiceV1;
+
+    private final NamingSubscriberServiceV2Impl namingSubscriberServiceV2;
+
+    public NamingSubscriberServiceLocalImpl(NamingSubscriberServiceV1Impl namingSubscriberServiceV1,
+            NamingSubscriberServiceV2Impl namingSubscriberServiceV2) {
+        this.namingSubscriberServiceV1 = namingSubscriberServiceV1;
+        this.namingSubscriberServiceV2 = namingSubscriberServiceV2;
+    }
+
+    @Override
+    public Collection<Subscriber> getSubscribers(String namespaceId, String serviceName) {
+        Collection<Subscriber> result = new HashSet<>();
+        result.addAll(namingSubscriberServiceV1.getSubscribers(namespaceId, serviceName));
+        result.addAll(namingSubscriberServiceV2.getSubscribers(namespaceId, serviceName));
+        return result;
+    }
+
+    @Override
+    public Collection<Subscriber> getSubscribers(Service service) {
+        Collection<Subscriber> result = new HashSet<>();
+        result.addAll(namingSubscriberServiceV1.getSubscribers(service));
+        result.addAll(namingSubscriberServiceV2.getSubscribers(service));
+        return result;
+    }
+
+    @Override
+    public Collection<Subscriber> getFuzzySubscribers(String namespaceId, String serviceName) {
+        Collection<Subscriber> result = new HashSet<>();
+        result.addAll(namingSubscriberServiceV1.getFuzzySubscribers(namespaceId, serviceName));
+        result.addAll(namingSubscriberServiceV2.getFuzzySubscribers(namespaceId, serviceName));
+        return result;
+    }
+
+    @Override
+    public Collection<Subscriber> getFuzzySubscribers(Service service) {
+        Collection<Subscriber> result = new HashSet<>();
+        result.addAll(namingSubscriberServiceV1.getFuzzySubscribers(service));
+        result.addAll(namingSubscriberServiceV2.getFuzzySubscribers(service));
+        return result;
+    }
+}
diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/UdpPushService.java b/naming/src/main/java/com/alibaba/nacos/naming/push/UdpPushService.java
@@ -26,6 +26,10 @@
 import com.alibaba.nacos.naming.misc.UtilsAndCommons;
 import com.alibaba.nacos.naming.monitor.MetricsMonitor;
 import com.alibaba.nacos.naming.pojo.Subscriber;
+import com.alibaba.nacos.naming.push.v1.ClientInfo;
+import com.alibaba.nacos.naming.push.v1.NamingSubscriberServiceV1Impl;
+import com.alibaba.nacos.naming.push.v1.PushClient;
+import com.alibaba.nacos.naming.push.v1.ServiceChangeEvent;
 import com.alibaba.nacos.naming.remote.udp.AckEntry;
 import com.alibaba.nacos.naming.remote.udp.AckPacket;
 import com.alibaba.nacos.naming.remote.udp.UdpConnector;

diff --git a/...alibaba/nacos/naming/push/ClientInfo.java → ...baba/nacos/naming/push/v1/ClientInfo.java b/...alibaba/nacos/naming/push/ClientInfo.java → ...baba/nacos/naming/push/v1/ClientInfo.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package com.alibaba.nacos.naming.push;
+package com.alibaba.nacos.naming.push.v1;
 
 import com.alibaba.nacos.naming.misc.UtilsAndCommons;
 import org.apache.commons.lang3.StringUtils;

diff --git a/...alibaba/nacos/naming/push/DataSource.java → ...baba/nacos/naming/push/v1/DataSource.java b/...alibaba/nacos/naming/push/DataSource.java → ...baba/nacos/naming/push/v1/DataSource.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package com.alibaba.nacos.naming.push;
+package com.alibaba.nacos.naming.push.v1;
 
 /**
  * Data source for naming push.

diff --git a/...g/push/NamingSubscriberServiceV1Impl.java → ...ush/v1/NamingSubscriberServiceV1Impl.java b/...g/push/NamingSubscriberServiceV1Impl.java → ...ush/v1/NamingSubscriberServiceV1Impl.java
@@ -14,14 +14,15 @@
  * limitations under the License.
  */
 
-package com.alibaba.nacos.naming.push;
+package com.alibaba.nacos.naming.push.v1;
 
 import com.alibaba.nacos.api.naming.utils.NamingUtils;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.misc.GlobalExecutor;
 import com.alibaba.nacos.naming.misc.Loggers;
 import com.alibaba.nacos.naming.misc.UtilsAndCommons;
 import com.alibaba.nacos.naming.pojo.Subscriber;
+import com.alibaba.nacos.naming.push.NamingSubscriberService;
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
@@ -36,8 +37,10 @@
  * Naming subscriber service for v1.x.
  *
  * @author xiweng.yy
+ * @deprecated Will be removed in v2.1.x version
  */
 @org.springframework.stereotype.Service
+@Deprecated
 public class NamingSubscriberServiceV1Impl implements NamingSubscriberService {
 
     private final ConcurrentMap<String, ConcurrentMap<String, PushClient>> clientMap = new ConcurrentHashMap<>();

diff --git a/...alibaba/nacos/naming/push/PushClient.java → ...baba/nacos/naming/push/v1/PushClient.java b/...alibaba/nacos/naming/push/PushClient.java → ...baba/nacos/naming/push/v1/PushClient.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package com.alibaba.nacos.naming.push;
+package com.alibaba.nacos.naming.push.v1;
 
 import com.alibaba.nacos.naming.misc.SwitchDomain;
 import com.alibaba.nacos.sys.utils.ApplicationUtils;