Merge pull request apache#467 from wu-sheng/fixed/438

wu-sheng · web-flow · commit 25036048aeb6 · 2017-09-27T15:16:33.000+08:00
Fixed apache#438, catch null point exception when collector start up with s…
diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneDataMonitor.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneDataMonitor.java
@@ -0,0 +1,71 @@
+package org.skywalking.apm.collector.cluster.standalone;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.client.h2.H2Client;
+import org.skywalking.apm.collector.client.zookeeper.util.PathUtils;
+import org.skywalking.apm.collector.core.CollectorException;
+import org.skywalking.apm.collector.core.client.Client;
+import org.skywalking.apm.collector.core.client.ClientException;
+import org.skywalking.apm.collector.core.client.DataMonitor;
+import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
+import org.skywalking.apm.collector.core.module.ModuleRegistration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ClusterStandaloneDataMonitor implements DataMonitor {
+
+    private final Logger logger = LoggerFactory.getLogger(ClusterStandaloneDataMonitor.class);
+
+    private H2Client client;
+
+    private Map<String, ClusterDataListener> listeners;
+    private Map<String, ModuleRegistration> registrations;
+
+    public ClusterStandaloneDataMonitor() {
+        listeners = new LinkedHashMap<>();
+        registrations = new LinkedHashMap<>();
+    }
+
+    @Override public void setClient(Client client) {
+        this.client = (H2Client)client;
+    }
+
+    @Override
+    public void addListener(ClusterDataListener listener, ModuleRegistration registration) throws ClientException {
+        String path = PathUtils.convertKey2Path(listener.path());
+        logger.info("listener path: {}", path);
+        listeners.put(path, listener);
+        registrations.put(path, registration);
+    }
+
+    @Override public ClusterDataListener getListener(String path) {
+        path = PathUtils.convertKey2Path(path);
+        return listeners.get(path);
+    }
+
+    @Override public void createPath(String path) throws ClientException {
+
+    }
+
+    @Override public void setData(String path, String value) throws ClientException {
+        if (listeners.containsKey(path)) {
+            listeners.get(path).addAddress(value);
+            listeners.get(path).serverJoinNotify(value);
+        }
+    }
+
+    @Override public void start() throws CollectorException {
+        Iterator<Map.Entry<String, ModuleRegistration>> entryIterator = registrations.entrySet().iterator();
+        while (entryIterator.hasNext()) {
+            Map.Entry<String, ModuleRegistration> next = entryIterator.next();
+            ModuleRegistration.Value value = next.getValue().buildValue();
+            String contextPath = value.getContextPath() == null ? "" : value.getContextPath();
+            setData(next.getKey(), value.getHostPort() + contextPath);
+        }
+    }
+}
diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java
@@ -15,6 +15,12 @@ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine {
 
     public static final String MODULE_NAME = "standalone";
 
+    private final ClusterStandaloneDataMonitor dataMonitor;
+
+    public ClusterStandaloneModuleDefine() {
+        this.dataMonitor = new ClusterStandaloneDataMonitor();
+    }
+
     @Override public String group() {
         return ClusterModuleGroupDefine.GROUP_NAME;
     }
@@ -32,14 +38,14 @@ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine {
     }
 
     @Override public DataMonitor dataMonitor() {
-        return null;
+        return dataMonitor;
     }
 
     @Override protected Client createClient() {
         return new H2Client();
     }
 
     @Override public ClusterModuleRegistrationReader registrationReader() {
-        return null;
+        return new ClusterStandaloneModuleRegistrationReader(dataMonitor);
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,12 @@ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine {`
`15`	`15`
`16`	`16`	`public static final String MODULE_NAME = "standalone";`
`17`	`17`
	`18`	`+ private final ClusterStandaloneDataMonitor dataMonitor;`
	`19`	`+`
	`20`	`+ public ClusterStandaloneModuleDefine() {`
	`21`	`+ this.dataMonitor = new ClusterStandaloneDataMonitor();`
	`22`	`+ }`
	`23`	`+`
`18`	`24`	`@Override public String group() {`
`19`	`25`	`return ClusterModuleGroupDefine.GROUP_NAME;`
`20`	`26`	`}`
`@@ -32,14 +38,14 @@ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine {`
`32`	`38`	`}`
`33`	`39`
`34`	`40`	`@Override public DataMonitor dataMonitor() {`
`35`		`- return null;`
	`41`	`+ return dataMonitor;`
`36`	`42`	`}`
`37`	`43`
`38`	`44`	`@Override protected Client createClient() {`
`39`	`45`	`return new H2Client();`
`40`	`46`	`}`
`41`	`47`
`42`	`48`	`@Override public ClusterModuleRegistrationReader registrationReader() {`
`43`		`- return null;`
	`49`	`+ return new ClusterStandaloneModuleRegistrationReader(dataMonitor);`
`44`	`50`	`}`
`45`	`51`	`}`