Merge pull request #1297 from Nicholas2015/feature_get_subscribers

Feature get subscribers

naming/src/main/java/com/alibaba/nacos/naming/controllers/ServiceController.java

@@ -28,6 +28,7 @@ import com.alibaba.nacos.naming.core.*;
 import com.alibaba.nacos.naming.exception.NacosException;
 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.selector.LabelSelector;
 import com.alibaba.nacos.naming.selector.NoneSelector;
 import com.alibaba.nacos.naming.selector.Selector;
@@ -40,7 +41,6 @@ import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.servlet.http.HttpServletRequest;
-import java.io.IOException;
 import java.net.URLDecoder;
 import java.util.*;
 
@@ -62,6 +62,9 @@ public class ServiceController {
     @Autowired
     private ServerListManager serverListManager;
 
+    @Autowired
+    private SubscribeManager subscribeManager;
+
     @RequestMapping(value = "", method = RequestMethod.POST)
     public String create(HttpServletRequest request) throws Exception {
 
@@ -365,6 +368,31 @@ public class ServiceController {
         return result;
     }
 
+    /**
+     * get subscriber list
+     *
+     * @param request
+     * @return
+     */
+    @RequestMapping(value = "/subscribers", method = RequestMethod.GET)
+    public JSONObject subscribers(HttpServletRequest request) {
+        String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID,
+            Constants.DEFAULT_NAMESPACE_ID);
+        String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
+        boolean aggregation = Boolean.valueOf(WebUtils.optional(request, "aggregation", String.valueOf(Boolean.TRUE)));
+
+        JSONObject result = new JSONObject();
+
+        try {
+            List<Subscriber> subscribers = subscribeManager.getSubscribers(serviceName, namespaceId, aggregation);
+            result.put("subscribers", subscribers);
+            return result;
+        } catch (InterruptedException e) {
+
+        }
+        return result;
+    }
+
     private List<String> filterInstanceMetadata(String namespaceId, List<String> serviceNames, String key, String value) {
 
         List<String> filteredServiceNames = new ArrayList<>();

naming/src/main/java/com/alibaba/nacos/naming/core/SubscribeManager.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 1999-2018 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.core;
+
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.nacos.api.naming.CommonParams;
+import com.alibaba.nacos.naming.boot.RunningConfig;
+import com.alibaba.nacos.naming.cluster.ServerListManager;
+import com.alibaba.nacos.naming.cluster.servers.Server;
+import com.alibaba.nacos.naming.misc.HttpClient;
+import com.alibaba.nacos.naming.misc.Loggers;
+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.PushService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.net.HttpURLConnection;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+/**
+ * @author Nicholas
+ * @since 1.0.1
+ */
+@Service
+public class SubscribeManager {
+
+    private static final String SUBSCRIBER_ON_SYNC_URL = "/service/subscribers";
+
+    @Autowired
+    private PushService pushService;
+
+    @Autowired
+    private ServerListManager serverListManager;
+
+
+    private List<Subscriber> getSubscribers(String serviceName, String namespaceId) {
+        return pushService.getClients(serviceName, namespaceId);
+    }
+
+    /**
+     * @param serviceName
+     * @param namespaceId
+     * @param aggregation
+     * @return
+     * @throws InterruptedException
+     */
+    public List<Subscriber> getSubscribers(String serviceName, String namespaceId, boolean aggregation) throws InterruptedException {
+        if (aggregation) {
+            // size = 1 means only myself in the list, we need at least one another server alive:
+            if (serverListManager.getHealthyServers().size() <= 1) {
+                return getSubscribers(serviceName, namespaceId);
+            }
+
+            List<Subscriber> subscriberList = new ArrayList<Subscriber>();
+            // try sync data from remote server:
+            for (Server server : serverListManager.getHealthyServers()) {
+
+                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.getKey())) {
+                    subscriberList.addAll(getSubscribers(serviceName, namespaceId));
+                }
+
+                HttpClient.HttpResult result = HttpClient.httpGet("http://" + server.getKey() + RunningConfig.getContextPath()
+                    + UtilsAndCommons.NACOS_NAMING_CONTEXT + SUBSCRIBER_ON_SYNC_URL, new ArrayList<>(), paramValues);
+
+                if (HttpURLConnection.HTTP_OK == result.code) {
+                    Subscribers subscribers = (Subscribers) JSONObject.parseObject(result.content, Subscribers.class);
+                    subscriberList.addAll(subscribers.getSubscribers());
+                }
+                return subscriberList.stream().filter(distinctByKey(Subscriber::toString)).collect(Collectors.toList());
+
+            }
+        } else {
+            // local server
+            return getSubscribers(serviceName, namespaceId);
+        }
+        return Collections.emptyList();
+    }
+
+    public static <T> Predicate<T> distinctByKey(Function<? super T, Object> keyExtractor) {
+        Map<Object, Boolean> seen = new ConcurrentHashMap<>(128);
+        return object -> seen.putIfAbsent(keyExtractor.apply(object), Boolean.TRUE) == null;
+    }
+}

naming/src/main/java/com/alibaba/nacos/naming/pojo/Subscriber.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 1999-2018 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.pojo;
+
+import java.io.Serializable;
+
+/**
+ * @author nicholas
+ * @version $Id: Subscriber.java, v 0.1 2019-05-28 下午10:47 nicholas Exp $$
+ */
+public class Subscriber implements Serializable {
+
+    private String addrStr;
+
+    private String agent;
+
+    private String app;
+
+    private String ip;
+
+    private String namespaceId;
+
+    private String serviceName;
+
+    public Subscriber(String addrStr, String agent, String app, String ip, String namespaceId, String serviceName) {
+        this.addrStr = addrStr;
+        this.agent = agent;
+        this.app = app;
+        this.ip = ip;
+        this.namespaceId = namespaceId;
+        this.serviceName = serviceName;
+    }
+
+    public String getAddrStr() {
+        return addrStr;
+    }
+
+    public void setAddrStr(String addrStr) {
+        this.addrStr = addrStr;
+    }
+
+    public String getAgent() {
+        return agent;
+    }
+
+    public void setAgent(String agent) {
+        this.agent = agent;
+    }
+
+    public String getApp() {
+        return app;
+    }
+
+    public void setApp(String app) {
+        this.app = app;
+    }
+
+    public String getIp() {
+        return ip;
+    }
+
+    public void setIp(String ip) {
+        this.ip = ip;
+    }
+
+    public String getNamespaceId() {
+        return namespaceId;
+    }
+
+    public void setNamespaceId(String namespaceId) {
+        this.namespaceId = namespaceId;
+    }
+
+    public String getServiceName() {
+        return serviceName;
+    }
+
+    public void setServiceName(String serviceName) {
+        this.serviceName = serviceName;
+    }
+}

naming/src/main/java/com/alibaba/nacos/naming/pojo/Subscribers.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 1999-2018 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.pojo;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * @author nicholas
+ * @version $Id: Subscribers.java, v 0.1 2019-05-28 下午10:47 nicholas Exp $$
+ */
+public class Subscribers implements Serializable {
+
+    private List<Subscriber> subscribers;
+
+    public List<Subscriber> getSubscribers() {
+        return subscribers;
+    }
+
+    public void setSubscribers(List<Subscriber> subscribers) {
+        this.subscribers = subscribers;
+    }
+}

naming/src/main/java/com/alibaba/nacos/naming/push/PushService.java

@@ -19,6 +19,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.nacos.naming.misc.Loggers;
 import com.alibaba.nacos.naming.misc.SwitchDomain;
 import com.alibaba.nacos.naming.misc.UtilsAndCommons;
+import com.alibaba.nacos.naming.pojo.Subscriber;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.codehaus.jackson.util.VersionUtil;
@@ -133,7 +134,7 @@ public class PushService {
                                  String tenant,
                                  String app) {
 
-        PushClient client = new PushService.PushClient(namespaceId,
+        PushClient client = new PushClient(namespaceId,
                 serviceName,
                 clusters,
                 agent,
@@ -166,6 +167,19 @@ public class PushService {
         }
     }
 
+    public List<Subscriber> getClients(String serviceName, String namespaceId) {
+        String serviceKey = UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName);
+        ConcurrentMap<String, PushClient> clientConcurrentMap = clientMap.get(serviceKey);
+        if (Objects.isNull(clientConcurrentMap)) {
+            return null;
+        }
+        List<Subscriber> clients = new ArrayList<Subscriber>();
+        clientConcurrentMap.forEach((key, client) -> {
+            clients.add(new Subscriber(client.getAddrStr(),client.getAgent(),client.getApp(),client.getIp(),namespaceId,serviceName));
+        });
+        return clients;
+    }
+
     public static void removeClientIfZombie() {
 
         int size = 0;

naming/src/test/java/com/alibaba/nacos/naming/core/SubscribeManagerTest.java

@@ -0,0 +1,97 @@
+package com.alibaba.nacos.naming.core;
+
+import com.alibaba.nacos.naming.BaseTest;
+import com.alibaba.nacos.naming.cluster.ServerListManager;
+import com.alibaba.nacos.naming.cluster.servers.Server;
+import com.alibaba.nacos.naming.pojo.Subscriber;
+import com.alibaba.nacos.naming.push.PushService;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Nicholas
+ */
+@SpringBootTest
+@RunWith(SpringRunner.class)
+public class SubscribeManagerTest extends BaseTest {
+
+    @Mock
+    private SubscribeManager subscribeManager;
+
+    @Mock
+    private PushService pushService;
+
+    @Mock
+    private ServerListManager serverListManager;
+
+    @Before
+    public void before() {
+        super.before();
+        subscribeManager = new SubscribeManager();
+    }
+
+    @Test
+    public void getSubscribersWithFalse() {
+        String serviceName = "test";
+        String namespaceId = "public";
+        boolean aggregation = Boolean.FALSE;
+        try {
+            List<Subscriber> clients = new ArrayList<Subscriber>();
+            Subscriber subscriber = new Subscriber("127.0.0.1:8080", "test", "app", "127.0.0.1", namespaceId, serviceName);
+            clients.add(subscriber);
+            Mockito.when(pushService.getClients(Mockito.anyString(), Mockito.anyString())).thenReturn(clients);
+            List<Subscriber> list = subscribeManager.getSubscribers(serviceName, namespaceId, aggregation);
+            Assert.assertNotNull(list);
+            Assert.assertEquals(1, list.size());
+            Assert.assertEquals("public", list.get(0).getNamespaceId());
+        } catch (Exception e) {
+
+        }
+    }
+
+    @Test
+    public void getSubscribersWithTrue() {
+        String serviceName = "test";
+        String namespaceId = "public";
+        boolean aggregation = Boolean.TRUE;
+        try {
+            List<Subscriber> clients = new ArrayList<Subscriber>();
+            Subscriber subscriber = new Subscriber("127.0.0.1:8080", "test", "app", "127.0.0.1", namespaceId, serviceName);
+            clients.add(subscriber);
+
+            List<Server> healthyServers = new ArrayList<>();
+
+            for (int i = 0; i <= 2; i++) {
+                Server server = new Server();
+                server.setIp("127.0.0.1");
+                server.setServePort(8080 + i);
+                server.setAlive(Boolean.TRUE);
+                server.setAdWeight(10);
+                server.setLastRefTime(System.currentTimeMillis());
+                server.setLastRefTimeStr(String.valueOf(System.currentTimeMillis()));
+                server.setSite("site");
+                server.setWeight(1);
+                healthyServers.add(server);
+            }
+
+            Mockito.when(serverListManager.getHealthyServers()).thenReturn(healthyServers);
+            //Mockito.doReturn(3).when(serverListManager.getHealthyServers().size());
+            List<Subscriber> list = subscribeManager.getSubscribers(serviceName, namespaceId, aggregation);
+            Assert.assertNotNull(list);
+            Assert.assertEquals(2, list.size());
+            Assert.assertEquals("public", list.get(0).getNamespaceId());
+        } catch (Exception e) {
+
+        }
+    }
+}
+

naming/src/test/java/com/alibaba/nacos/naming/pojo/SubscriberTest.java

@@ -0,0 +1,46 @@
+package com.alibaba.nacos.naming.pojo;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Nicholas
+ */
+public class SubscriberTest {
+
+    @Test
+    public void subscriberBeanTest() {
+        Subscriber subscriber = new Subscriber("127.0.0.1:8080", "agent", "app", "127.0.0.1", "public", "test");
+        subscriber.setAddrStr("127.0.0.1:8080");
+        subscriber.setIp("127.0.0.1");
+        subscriber.setApp("app");
+        subscriber.setAgent("agent");
+        subscriber.setNamespaceId("public");
+        subscriber.setServiceName("test");
+
+        subscriber.getAddrStr();
+        subscriber.getIp();
+        subscriber.getAgent();
+        subscriber.getApp();
+        subscriber.getNamespaceId();
+        subscriber.getServiceName();
+
+        Subscribers subscribers = new Subscribers();
+        List<Subscriber> subscriberList = new ArrayList<>();
+        subscriberList.add(subscriber);
+        subscribers.setSubscribers(subscriberList);
+        subscribers.getSubscribers();
+
+        Assert.assertNotNull(subscriberList);
+        Assert.assertEquals(1, subscriberList.size());
+        Assert.assertEquals("127.0.0.1:8080", subscriberList.get(0).getAddrStr());
+        Assert.assertEquals("127.0.0.1", subscriberList.get(0).getIp());
+        Assert.assertEquals("app", subscriberList.get(0).getApp());
+        Assert.assertEquals("agent", subscriberList.get(0).getAgent());
+        Assert.assertEquals("public", subscriberList.get(0).getNamespaceId());
+        Assert.assertEquals("test", subscriberList.get(0).getServiceName());
+    }
+}