注册中心添加stop功能

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/RegistryNodeManager.java

@@ -34,4 +34,8 @@ public class RegistryNodeManager {
     public List<String> getConnectionURLOfPreviousChanged() {
         return new ArrayList<String>(connectionURLOfPreviousChanged);
     }
+
+    public void clear() {
+        connectionURLOfPreviousChanged = new ArrayList<String>();
+    }
 }

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/api/RegistryCenter.java

@@ -30,4 +30,7 @@ public interface RegistryCenter {
      * @param centerConfig 配置参数
      */
     void start(Properties centerConfig);
+
+
+    void stop();
 }

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/impl/zookeeper/ZookeeperRegistryCenter.java

@@ -59,6 +59,10 @@ public class ZookeeperRegistryCenter implements RegistryCenter {
         }
     }
 
+    @Override
+    public void stop() {
+        nodeManager.clear();
+    }
 
 
 }

skywalking-commons/skywalking-registry/src/test/java/com/a/eye/skywalking/registry/impl/zookeeper/ZookeeperRegistryCenterTest.java

@@ -40,6 +40,24 @@ public class ZookeeperRegistryCenterTest extends TestSuite {
     public void tearDown() throws Exception {
         zkTestServer.getTempDirectory().delete();
         zkTestServer.stop();
+        registryCenter.stop();
+    }
+
+    @Test
+    public void subscribeNodeTest() throws InterruptedException {
+        final StringBuilder addUrl = new StringBuilder();
+        registryCenter.subscribe("/skywalking/storage",  new NotifyListener() {
+            @Override
+            public void notify(List<RegistryNode> registryNodes) {
+                for (RegistryNode url : registryNodes) {
+                    addUrl.append(url.getNode() + ",");
+                }
+            }
+        });
+
+        registryCenter.register("/skywalking/storage/127.0.0.1:9400");
+        Thread.sleep(100L);
+        assertEquals(addUrl.deleteCharAt(addUrl.length() - 1).toString(), "127.0.0.1:9400");
     }
 
 
@@ -59,6 +77,8 @@ public class ZookeeperRegistryCenterTest extends TestSuite {
         assertEquals(addUrl.deleteCharAt(addUrl.length() - 1).toString(), "127.0.0.1:9400");
     }
 
+
+
     @Test
     public void registryNodeTest() throws IOException, InterruptedException, KeeperException {
         registryCenter.register("/skywalking/storage/test");

skywalking-storage-center/skywalking-routing/src/test/java/com/a/eye/skywalking/routing/router/NodeChangesListenerTest.java

-package com.a.eye.skywalking.routing.router;
-
-import com.a.eye.skywalking.registry.RegistryCenterFactory;
-import com.a.eye.skywalking.registry.api.CenterType;
-import com.a.eye.skywalking.registry.api.RegistryCenter;
-import com.a.eye.skywalking.registry.impl.zookeeper.ZookeeperConfig;
-import com.a.eye.skywalking.routing.config.Config;
-import org.apache.curator.test.TestingServer;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Spy;
-import org.mockito.runners.MockitoJUnitRunner;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Properties;
-
-@RunWith(MockitoJUnitRunner.class)
-public class NodeChangesListenerTest {
-
-    @Spy
-    private Router router = new Router();
-
-    private TestingServer zkTestServer;
-    private RegistryCenter registryCenter;
-
-    @Before
-    public void setUp() throws Exception {
-        zkTestServer = new TestingServer(2181, true);
-        registryCenter = RegistryCenterFactory.INSTANCE.getRegistryCenter(CenterType.DEFAULT_CENTER_TYPE);
-        Properties config = new Properties();
-        config.put(ZookeeperConfig.CONNECT_URL, "127.0.0.1:2181");
-        registryCenter.start(config);
-
-    }
-
-    @Test
-    public void testRoutingStartBeforeStorageNode() throws InterruptedException {
-        registryCenter.register(Config.StorageNode.SUBSCRIBE_PATH + "/127.0.0.1:34000");
-        Thread.sleep(10);
-        List<String> nodeURL = new ArrayList<>();
-        nodeURL.add("127.0.0.1:34000");
-        //verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
-    }
-
-
-    @Test
-    public void testStorageNodeStartBeforeRoutingStart() throws InterruptedException {
-        registryCenter.register(Config.StorageNode.SUBSCRIBE_PATH + "/127.0.0.1:34000");
-        Thread.sleep(10);
-        List<String> nodeURL = new ArrayList<>();
-        nodeURL.add("127.0.0.1:34000");
-        //verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
-    }
-
-    @After
-    public void tearUp() throws IOException {
-        zkTestServer.stop();
-    }
-
-}