Merge pull request #1715 from rushsky518/develop

receive config change event

api/src/main/java/com/alibaba/nacos/api/common/Constants.java

@@ -62,6 +62,8 @@ public class Constants {
 
     public static final String CONFIG_VERSION = "Config-Version";
 
+    public static final String CONFIG_TYPE = "Config-Type";
+
     public static final String IF_MODIFIED_SINCE = "If-Modified-Since";
 
     public static final String SPACING_INTERVAL = "client-spacing-interval";

api/src/main/java/com/alibaba/nacos/api/config/ConfigChangeEvent.java

@@ -0,0 +1,44 @@
+/*
+ * 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.api.config;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * ConfigChangeEvent
+ *
+ * @author rushsky518
+ */
+public class ConfigChangeEvent {
+    private Map<String, ConfigChangeItem> data;
+
+    public ConfigChangeEvent(Map<String, ConfigChangeItem> data) {
+        this.data = data;
+    }
+
+    public ConfigChangeItem getChangeItem(String key) {
+        return data.get(key);
+    }
+
+    public Collection<ConfigChangeItem> getChangeItems() {
+        return data.values();
+    }
+
+}
+

api/src/main/java/com/alibaba/nacos/api/config/ConfigChangeItem.java

@@ -0,0 +1,77 @@
+/*
+ * 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.api.config;
+
+/**
+ * ConfigChangeItem
+ *
+ * @author rushsky518
+ */
+public class ConfigChangeItem {
+    private String key;
+    private String oldValue;
+    private String newValue;
+
+    private PropertyChangeType type;
+
+    public ConfigChangeItem(String key, String oldValue, String newValue) {
+        this.key = key;
+        this.oldValue = oldValue;
+        this.newValue = newValue;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public String getOldValue() {
+        return oldValue;
+    }
+
+    public void setOldValue(String oldValue) {
+        this.oldValue = oldValue;
+    }
+
+    public String getNewValue() {
+        return newValue;
+    }
+
+    public void setNewValue(String newValue) {
+        this.newValue = newValue;
+    }
+
+    public PropertyChangeType getType() {
+        return type;
+    }
+
+    public void setType(PropertyChangeType type) {
+        this.type = type;
+    }
+
+    @Override
+    public String toString() {
+        return "ConfigChangeItem{" +
+            "key='" + key + '\'' +
+            ", oldValue='" + oldValue + '\'' +
+            ", newValue='" + newValue + '\'' +
+            ", type=" + type +
+            '}';
+    }
+}

api/src/main/java/com/alibaba/nacos/api/config/PropertyChangeType.java

@@ -0,0 +1,30 @@
+/*
+ * 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.api.config;
+
+/**
+ * Property Change Type
+ *
+ * @author rushsky518
+ */
+public enum PropertyChangeType {
+    /** add */
+    ADDED,
+    /** modified */
+    MODIFIED,
+    /** deleted */
+    DELETED
+}

api/src/main/java/com/alibaba/nacos/api/config/listener/ConfigChangeParser.java

@@ -0,0 +1,45 @@
+/*
+ * 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.api.config.listener;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * ConfigChangeParser
+ *
+ * @author rushsky518
+ */
+public interface ConfigChangeParser {
+    /**
+     * judge type
+     * @param type
+     * @return
+     */
+    boolean isResponsibleFor(String type);
+
+    /**
+     * compare old and new data
+     * @param oldContent
+     * @param newContent
+     * @param type
+     * @return
+     * @throws IOException
+     */
+    Map<String, ConfigChangeItem> doParse(String oldContent, String newContent, String type) throws IOException;
+}

client/pom.xml

@@ -116,6 +116,10 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+        </dependency>
 
     </dependencies>
 

client/src/main/java/com/alibaba/nacos/client/config/NacosConfigService.java

@@ -170,9 +170,8 @@ public class NacosConfigService implements ConfigService {
         }
 
         try {
-            content = worker.getServerConfig(dataId, group, tenant, timeoutMs);
+            String[] ct = worker.getServerConfig(dataId, group, tenant, timeoutMs);
-
+            cr.setContent(ct[0]);
-            cr.setContent(content);
 
             configFilterChainManager.doFilter(null, cr);
             content = cr.getContent();

client/src/main/java/com/alibaba/nacos/client/config/impl/AbstractConfigChangeParser.java

@@ -0,0 +1,74 @@
+/*
+ * 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.client.config.impl;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.api.config.PropertyChangeType;
+import com.alibaba.nacos.api.config.listener.ConfigChangeParser;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * AbstractConfigChangeParser
+ *
+ * @author rushsky518
+ */
+public abstract class AbstractConfigChangeParser implements ConfigChangeParser {
+    private String configType;
+
+    public AbstractConfigChangeParser(String configType) {
+        this.configType = configType;
+    }
+
+    @Override
+    public boolean isResponsibleFor(String type) {
+        return this.configType.equalsIgnoreCase(type);
+    }
+
+    protected Map<String, ConfigChangeItem> filterChangeData(Map oldMap, Map newMap) {
+        Map<String, ConfigChangeItem> result = new HashMap<String, ConfigChangeItem>(16);
+        for (Iterator<Map.Entry<String, Object>> entryItr = oldMap.entrySet().iterator(); entryItr.hasNext();) {
+            Map.Entry<String, Object> e = entryItr.next();
+            ConfigChangeItem cci = null;
+            if (newMap.containsKey(e.getKey()))  {
+                if (e.getValue().equals(newMap.get(e.getKey()))) {
+                    continue;
+                }
+                cci = new ConfigChangeItem(e.getKey(), e.getValue().toString(), newMap.get(e.getKey()).toString());
+                cci.setType(PropertyChangeType.MODIFIED);
+            } else {
+                cci = new ConfigChangeItem(e.getKey(), e.getValue().toString(), null);
+                cci.setType(PropertyChangeType.DELETED);
+            }
+
+            result.put(e.getKey(), cci);
+        }
+
+        for (Iterator<Map.Entry<String, Object>> entryItr = newMap.entrySet().iterator(); entryItr.hasNext();) {
+            Map.Entry<String, Object> e = entryItr.next();
+            if (!oldMap.containsKey(e.getKey())) {
+                ConfigChangeItem cci = new ConfigChangeItem(e.getKey(), null, e.getValue().toString());
+                cci.setType(PropertyChangeType.ADDED);
+                result.put(e.getKey(), cci);
+            }
+        }
+
+        return result;
+    }
+
+}

client/src/main/java/com/alibaba/nacos/client/config/impl/CacheData.java

@@ -16,11 +16,13 @@
 package com.alibaba.nacos.client.config.impl;
 
 import com.alibaba.nacos.api.common.Constants;
+import com.alibaba.nacos.api.config.ConfigChangeEvent;
 import com.alibaba.nacos.api.config.listener.AbstractSharedListener;
 import com.alibaba.nacos.api.config.listener.Listener;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.client.config.filter.impl.ConfigFilterChainManager;
 import com.alibaba.nacos.client.config.filter.impl.ConfigResponse;
+import com.alibaba.nacos.client.config.listener.impl.AbstractConfigChangeListener;
 import com.alibaba.nacos.client.config.utils.MD5;
 import com.alibaba.nacos.client.utils.LogUtils;
 import com.alibaba.nacos.client.utils.TenantUtil;
@@ -28,6 +30,7 @@ import org.slf4j.Logger;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
@@ -59,9 +62,17 @@ public class CacheData {
         return content;
     }
 
-    public void setContent(String newContent) {
+    public void setContent(String content) {
-        this.content = newContent;
+        this.content = content;
-        this.md5 = getMd5String(content);
+        this.md5 = getMd5String(this.content);
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
     }
 
     /**
@@ -74,7 +85,9 @@ public class CacheData {
         if (null == listener) {
             throw new IllegalArgumentException("listener is null");
         }
-        ManagerListenerWrap wrap = new ManagerListenerWrap(listener, md5);
+        ManagerListenerWrap wrap = (listener instanceof AbstractConfigChangeListener) ?
+            new ManagerListenerWrap(listener, md5, content) : new ManagerListenerWrap(listener, md5);
+
         if (listeners.addIfAbsent(wrap)) {
             LOGGER.info("[{}] [add-listener] ok, tenant={}, dataId={}, group={}, cnt={}", name, tenant, dataId, group,
                 listeners.size());
@@ -158,12 +171,12 @@ public class CacheData {
     void checkListenerMd5() {
         for (ManagerListenerWrap wrap : listeners) {
             if (!md5.equals(wrap.lastCallMd5)) {
-                safeNotifyListener(dataId, group, content, md5, wrap);
+                safeNotifyListener(dataId, group, content, type, md5, wrap);
             }
         }
     }
 
-    private void safeNotifyListener(final String dataId, final String group, final String content,
+    private void safeNotifyListener(final String dataId, final String group, final String content, final String type,
                                     final String md5, final ManagerListenerWrap listenerWrap) {
         final Listener listener = listenerWrap.listener;
 
@@ -188,6 +201,15 @@ public class CacheData {
                     configFilterChainManager.doFilter(null, cr);
                     String contentTmp = cr.getContent();
                     listener.receiveConfigInfo(contentTmp);
+
+                    // compare lastContent and content
+                    if (listener instanceof AbstractConfigChangeListener) {
+                        Map data = ConfigChangeHandler.getInstance().parseChangeData(listenerWrap.lastContent, content, type);
+                        ConfigChangeEvent event = new ConfigChangeEvent(data);
+                        ((AbstractConfigChangeListener)listener).receiveConfigChange(event);
+                        listenerWrap.lastContent = content;
+                    }
+
                     listenerWrap.lastCallMd5 = md5;
                     LOGGER.info("[{}] [notify-ok] dataId={}, group={}, md5={}, listener={} ", name, dataId, group, md5,
                         listener);
@@ -282,11 +304,13 @@ public class CacheData {
     private volatile String content;
     private int taskId;
     private volatile boolean isInitializing = true;
+    private String type;
 }
 
 class ManagerListenerWrap {
     final Listener listener;
     String lastCallMd5 = CacheData.getMd5String(null);
+    String lastContent = null;
 
     ManagerListenerWrap(Listener listener) {
         this.listener = listener;
@@ -297,6 +321,12 @@ class ManagerListenerWrap {
         this.lastCallMd5 = md5;
     }
 
+    ManagerListenerWrap(Listener listener, String md5, String lastContent) {
+        this.listener = listener;
+        this.lastCallMd5 = md5;
+        this.lastContent = lastContent;
+    }
+
     @Override
     public boolean equals(Object obj) {
         if (null == obj || obj.getClass() != getClass()) {

client/src/main/java/com/alibaba/nacos/client/config/impl/ClientWorker.java

@@ -17,6 +17,7 @@ package com.alibaba.nacos.client.config.impl;
 
 import com.alibaba.nacos.api.PropertyKeyConst;
 import com.alibaba.nacos.api.common.Constants;
+import com.alibaba.nacos.api.config.ConfigType;
 import com.alibaba.nacos.api.config.listener.Listener;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.client.config.common.GroupKey;
@@ -46,6 +47,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static com.alibaba.nacos.api.common.Constants.LINE_SEPARATOR;
 import static com.alibaba.nacos.api.common.Constants.WORD_SEPARATOR;
+import static com.alibaba.nacos.api.common.Constants.CONFIG_TYPE;
 
 /**
  * Longpolling
@@ -183,8 +185,8 @@ public class ClientWorker {
                 cache = new CacheData(configFilterChainManager, agent.getName(), dataId, group, tenant);
                 // fix issue # 1317
                 if (enableRemoteSyncConfig) {
-                    String content = getServerConfig(dataId, group, tenant, 3000L);
+                    String[] ct = getServerConfig(dataId, group, tenant, 3000L);
-                    cache.setContent(content);
+                    cache.setContent(ct[0]);
                 }
             }
 
@@ -210,8 +212,9 @@ public class ClientWorker {
         return cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
     }
 
-    public String getServerConfig(String dataId, String group, String tenant, long readTimeout)
+    public String[] getServerConfig(String dataId, String group, String tenant, long readTimeout)
         throws NacosException {
+        String[] ct = new String[2];
         if (StringUtils.isBlank(group)) {
             group = Constants.DEFAULT_GROUP;
         }
@@ -236,10 +239,16 @@ public class ClientWorker {
         switch (result.code) {
             case HttpURLConnection.HTTP_OK:
                 LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, result.content);
-                return result.content;
+                ct[0] = result.content;
+                if (result.headers.containsKey(CONFIG_TYPE)) {
+                    ct[1] = result.headers.get(CONFIG_TYPE).get(0);
+                } else {
+                    ct[1] = ConfigType.TEXT.getType();
+                }
+                return ct;
             case HttpURLConnection.HTTP_NOT_FOUND:
                 LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, null);
-                return null;
+                return ct;
             case HttpURLConnection.HTTP_CONFLICT: {
                 LOGGER.error(
                     "[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, "
@@ -517,12 +526,15 @@ public class ClientWorker {
                         tenant = key[2];
                     }
                     try {
-                        String content = getServerConfig(dataId, group, tenant, 3000L);
+                        String[] ct = getServerConfig(dataId, group, tenant, 3000L);
                         CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
-                        cache.setContent(content);
+                        cache.setContent(ct[0]);
-                        LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}",
+                        if (null != ct[1]) {
+                            cache.setType(ct[1]);
+                        }
+                        LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}, type={}",
                             agent.getName(), dataId, group, tenant, cache.getMd5(),
-                            ContentUtils.truncateContent(content));
+                            ContentUtils.truncateContent(ct[0]), ct[1]);
                     } catch (NacosException ioe) {
                         String message = String.format(
                             "[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s",

client/src/main/java/com/alibaba/nacos/client/config/impl/ConfigChangeHandler.java

@@ -0,0 +1,67 @@
+/*
+ * 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.client.config.impl;
+
+import com.alibaba.nacos.api.config.listener.ConfigChangeParser;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.ServiceLoader;
+import java.util.Map;
+import java.util.Iterator;
+
+/**
+ * ConfigChangeHandler
+ *
+ * @author rushsky518
+ */
+public class ConfigChangeHandler {
+    private static class ConfigChangeHandlerHolder {
+        private final static ConfigChangeHandler INSTANCE = new ConfigChangeHandler();
+    }
+
+    private ConfigChangeHandler() {
+        this.parserList = new LinkedList<ConfigChangeParser>();
+
+        ServiceLoader<ConfigChangeParser> loader = ServiceLoader.load(ConfigChangeParser.class);
+        Iterator<ConfigChangeParser> itr = loader.iterator();
+        while (itr.hasNext()) {
+            this.parserList.add(itr.next());
+        }
+
+        this.parserList.add(new PropertiesChangeParser());
+        this.parserList.add(new YmlChangeParser());
+    }
+
+    public static ConfigChangeHandler getInstance() {
+        return ConfigChangeHandlerHolder.INSTANCE;
+    }
+
+    public Map parseChangeData(String oldContent, String newContent, String type) throws IOException {
+        for (ConfigChangeParser changeParser: this.parserList) {
+            if (changeParser.isResponsibleFor(type)) {
+                return changeParser.doParse(oldContent, newContent, type);
+            }
+        }
+
+        return Collections.emptyMap();
+    }
+
+    private List<ConfigChangeParser> parserList;
+
+}

client/src/main/java/com/alibaba/nacos/client/config/impl/PropertiesChangeParser.java

@@ -0,0 +1,50 @@
+/*
+ * 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.client.config.impl;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.client.utils.StringUtils;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * PropertiesChangeParser
+ *
+ * @author rushsky518
+ */
+public class PropertiesChangeParser extends AbstractConfigChangeParser {
+    public PropertiesChangeParser() {
+        super("properties");
+    }
+
+    @Override
+    public Map<String, ConfigChangeItem> doParse(String oldContent, String newContent, String type) throws IOException {
+        Properties oldProps = new Properties();
+        Properties newProps = new Properties();
+
+        if (StringUtils.isNotBlank(oldContent)) {
+            oldProps.load(new StringReader(oldContent));
+        }
+        if (StringUtils.isNotBlank(newContent)) {
+            newProps.load(new StringReader(newContent));
+        }
+
+        return filterChangeData(oldProps, newProps);
+    }
+}

client/src/main/java/com/alibaba/nacos/client/config/impl/YmlChangeParser.java

@@ -0,0 +1,90 @@
+/*
+ * 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.client.config.impl;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.client.utils.StringUtils;
+import org.yaml.snakeyaml.Yaml;
+import java.util.*;
+
+/**
+ * YmlChangeParser
+ *
+ * @author rushsky518
+ */
+public class YmlChangeParser extends AbstractConfigChangeParser {
+    public YmlChangeParser() {
+        super("yaml");
+    }
+
+    @Override
+    public Map<String, ConfigChangeItem> doParse(String oldContent, String newContent, String type) {
+        Map<String, Object> oldMap = Collections.emptyMap();
+        Map<String, Object> newMap = Collections.emptyMap();
+
+        if (StringUtils.isNotBlank(oldContent)) {
+            oldMap = (new Yaml()).load(oldContent);
+            oldMap = getFlattenedMap(oldMap);
+        }
+        if (StringUtils.isNotBlank(newContent)) {
+            newMap = (new Yaml()).load(newContent);
+            newMap = getFlattenedMap(newMap);
+        }
+
+        return filterChangeData(oldMap, newMap);
+    }
+
+    private final Map<String, Object> getFlattenedMap(Map<String, Object> source) {
+        Map<String, Object> result = new LinkedHashMap<String, Object>(128);
+        buildFlattenedMap(result, source, null);
+        return result;
+    }
+
+    private void buildFlattenedMap(Map<String, Object> result, Map<String, Object> source, String path) {
+        for (Iterator<Map.Entry<String, Object>> itr = source.entrySet().iterator(); itr.hasNext(); ) {
+            Map.Entry<String, Object> e = itr.next();
+            String key = e.getKey();
+            if (StringUtils.isNotBlank(path)) {
+                if (e.getKey().startsWith("[")) {
+                    key = path + key;
+                } else {
+                    key = path + '.' + key;
+                }
+            }
+            if (e.getValue() instanceof String) {
+                result.put(key, e.getValue());
+            } else if (e.getValue() instanceof Map) {
+                @SuppressWarnings("unchecked")
+                Map<String, Object> map = (Map<String, Object>) e.getValue();
+                buildFlattenedMap(result, map, key);
+            } else if (e.getValue() instanceof Collection) {
+                @SuppressWarnings("unchecked")
+                Collection<Object> collection = (Collection<Object>) e.getValue();
+                if (collection.isEmpty()) {
+                    result.put(key, "");
+                } else {
+                    int count = 0;
+                    for (Object object : collection) {
+                        buildFlattenedMap(result, Collections.singletonMap("[" + (count++) + "]", object), key);
+                    }
+                }
+            } else {
+                result.put(key, (e.getValue() != null ? e.getValue() : ""));
+            }
+        }
+    }
+
+}

client/src/main/java/com/alibaba/nacos/client/config/listener/impl/AbstractConfigChangeListener.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.client.config.listener.impl;
+
+import com.alibaba.nacos.api.config.listener.AbstractListener;
+import com.alibaba.nacos.api.config.ConfigChangeEvent;
+
+/**
+ * AbstractConfigChangeListener
+ *
+ * @author rushsky518
+ */
+public abstract class AbstractConfigChangeListener extends AbstractListener {
+    /**
+     * handle config change
+     * @param event
+     */
+    public abstract void receiveConfigChange(final ConfigChangeEvent event);
+
+    @Override
+    public void receiveConfigInfo(final String configInfo) {}
+}
+

client/src/test/java/com/alibaba/nacos/client/config/listener/impl/ConfigChangeHandlerTest.java

@@ -0,0 +1,39 @@
+/*
+ * 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.client.config.listener.impl;
+
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.client.config.impl.ConfigChangeHandler;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class ConfigChangeHandlerTest {
+    @Test
+    public void testParseProperties() throws IOException {
+        Map properties = ConfigChangeHandler.getInstance().parseChangeData("", "app.name = nacos", "properties");
+        Assert.assertEquals("nacos", ((ConfigChangeItem)properties.get("app.name")).getNewValue());
+    }
+
+    @Test
+    public void testParseYaml() throws IOException {
+        Map properties = ConfigChangeHandler.getInstance().parseChangeData("", "app:\n  name: nacos", "yaml");
+        Assert.assertEquals("nacos", ((ConfigChangeItem)properties.get("app.name")).getNewValue());
+    }
+}

client/src/test/java/com/alibaba/nacos/client/config/listener/impl/PropertiesChangeParserTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.client.config.listener.impl;
+
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.client.config.impl.PropertiesChangeParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class PropertiesChangeParserTest {
+    private PropertiesChangeParser parser = new PropertiesChangeParser();
+    private final String type = "properties";
+
+    @Test
+    public void testType() {
+        Assert.assertEquals(true, parser.isResponsibleFor(type));
+    }
+
+    @Test
+    public void testAddKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("", "app.name = nacos", type);
+        Assert.assertEquals(null, map.get("app.name").getOldValue());
+        Assert.assertEquals("nacos", map.get("app.name").getNewValue());
+    }
+
+    @Test
+    public void testRemoveKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("app.name = nacos", "", type);
+        Assert.assertEquals("nacos", map.get("app.name").getOldValue());
+        Assert.assertEquals(null, map.get("app.name").getNewValue());
+    }
+
+    @Test
+    public void testModifyKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("app.name = rocketMQ", "app.name = nacos", type);
+        Assert.assertEquals("rocketMQ", map.get("app.name").getOldValue());
+        Assert.assertEquals("nacos", map.get("app.name").getNewValue());
+    }
+
+}
+

client/src/test/java/com/alibaba/nacos/client/config/listener/impl/YmlChangeParserTest.java

@@ -0,0 +1,57 @@
+/*
+ * 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.client.config.listener.impl;
+
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.client.config.impl.YmlChangeParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class YmlChangeParserTest {
+    private YmlChangeParser parser = new YmlChangeParser();
+    private final String type = "yaml";
+
+    @Test
+    public void testType() {
+        Assert.assertEquals(true, parser.isResponsibleFor(type));
+    }
+
+    @Test
+    public void testAddKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("", "app:\n  name: nacos", type);
+        Assert.assertEquals(null, map.get("app.name").getOldValue());
+        Assert.assertEquals("nacos", map.get("app.name").getNewValue());
+    }
+
+    @Test
+    public void testRemoveKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("app:\n  name: nacos", "", type);
+        Assert.assertEquals("nacos", map.get("app.name").getOldValue());
+        Assert.assertEquals(null, map.get("app.name").getNewValue());
+    }
+
+    @Test
+    public void testModifyKey() throws IOException {
+        Map<String, ConfigChangeItem> map = parser.doParse("app:\n  name: rocketMQ", "app:\n  name: nacos", type);
+        Assert.assertEquals("rocketMQ", map.get("app.name").getOldValue());
+        Assert.assertEquals("nacos", map.get("app.name").getNewValue());
+    }
+}
+

config/src/main/java/com/alibaba/nacos/config/server/controller/ConfigServletInner.java

@@ -130,6 +130,8 @@ public class ConfigServletInner {
                             isBeta = true;
                         }
                     }
+                    String configType = cacheItem.getType();
+                    response.setHeader("Config-Type", (null != configType) ? configType : "text");
                 }
                 File file = null;
                 ConfigInfoBase configInfoBase = null;

config/src/main/java/com/alibaba/nacos/config/server/model/CacheItem.java

@@ -109,6 +109,14 @@ public class CacheItem {
         this.tagLastModifiedTs = tagLastModifiedTs;
     }
 
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
     final String groupKey;
     public volatile String md5 = Constants.NULL;
     public volatile long lastModifiedTs;
@@ -123,5 +131,6 @@ public class CacheItem {
     public volatile Map<String, String> tagMd5;
     public volatile Map<String, Long> tagLastModifiedTs;
     public SimpleReadWriteLock rwLock = new SimpleReadWriteLock();
+    public String type;
 
 }

config/src/main/java/com/alibaba/nacos/config/server/model/ConfigInfo.java

@@ -28,6 +28,8 @@ public class ConfigInfo extends ConfigInfoBase {
 
     private String appName;
 
+    private String type;
+
     public ConfigInfo() {
 
     }
@@ -63,6 +65,14 @@ public class ConfigInfo extends ConfigInfoBase {
         this.appName = appName;
     }
 
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
     @Override
     public int hashCode() {
         return super.hashCode();

config/src/main/java/com/alibaba/nacos/config/server/service/ConfigService.java

@@ -57,9 +57,10 @@ public class ConfigService {
     /**
      * 保存配置文件，并缓存md5.
      */
-    static public boolean dump(String dataId, String group, String tenant, String content, long lastModifiedTs) {
+    static public boolean dump(String dataId, String group, String tenant, String content, long lastModifiedTs, String type) {
         String groupKey = GroupKey2.getKey(dataId, group, tenant);
-        makeSure(groupKey);
+        CacheItem ci = makeSure(groupKey);
+        ci.setType(type);
         final int lockResult = tryWriteLock(groupKey);
         assert (lockResult != 0);
 

config/src/main/java/com/alibaba/nacos/config/server/service/PersistService.java

@@ -113,6 +113,7 @@ public class PersistService {
             info.setGroup(rs.getString("group_id"));
             info.setTenant(rs.getString("tenant_id"));
             info.setAppName(rs.getString("app_name"));
+            info.setType(rs.getString("type"));
 
             try {
                 info.setContent(rs.getString("content"));
@@ -235,6 +236,11 @@ public class PersistService {
             } catch (SQLException e) {
                 // ignore
             }
+            try {
+                info.setType(rs.getString("type"));
+            } catch (SQLException e) {
+                // ignore
+            }
             return info;
         }
     }
@@ -1325,7 +1331,7 @@ public class PersistService {
         final String appName = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("appName");
         final String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
         String sqlCount = "select count(*) from config_info";
-        String sql = "select ID,data_id,group_id,tenant_id,app_name,content from config_info";
+        String sql = "select ID,data_id,group_id,tenant_id,app_name,content,type from config_info";
         StringBuilder where = new StringBuilder(" where ");
         List<String> paramList = new ArrayList<String>();
         paramList.add(tenantTmp);
@@ -1944,7 +1950,7 @@ public class PersistService {
 
     public Page<ConfigInfoWrapper> findAllConfigInfoFragment(final long lastMaxId, final int pageSize) {
         String select
-            = "SELECT id,data_id,group_id,tenant_id,app_name,content,md5,gmt_modified from config_info where id > ? "
+            = "SELECT id,data_id,group_id,tenant_id,app_name,content,md5,gmt_modified,type from config_info where id > ? "
             + "order by id asc limit ?,?";
         PaginationHelper<ConfigInfoWrapper> helper = new PaginationHelper<ConfigInfoWrapper>();
         try {
@@ -2927,8 +2933,8 @@ public class PersistService {
         final String tenantTmp = StringUtils.isBlank(tenant) ? StringUtils.EMPTY : tenant;
         try {
             return this.jt.queryForObject(
-                "SELECT ID,data_id,group_id,tenant_id,app_name,content,md5 FROM config_info WHERE data_id=? AND group_id=? AND tenant_id=?",
+                "SELECT ID,data_id,group_id,tenant_id,app_name,content,md5,type FROM config_info WHERE data_id=? AND group_id=? AND tenant_id=?",
-                new Object[]{dataId, group, tenantTmp}, CONFIG_INFO_ROW_MAPPER);
+                new Object[] {dataId, group, tenantTmp}, CONFIG_INFO_ROW_MAPPER);
         } catch (EmptyResultDataAccessException e) { // 表明数据不存在, 返回null
             return null;
         } catch (CannotGetJdbcConnectionException e) {

config/src/main/java/com/alibaba/nacos/config/server/service/dump/DumpTask.java

@@ -187,7 +187,7 @@ class DumpProcessor implements TaskProcessor {
 
                 boolean result;
                 if (null != cf) {
-                    result = ConfigService.dump(dataId, group, tenant, cf.getContent(), lastModified);
+                    result = ConfigService.dump(dataId, group, tenant, cf.getContent(), lastModified, cf.getType());
 
                     if (result) {
                         ConfigTraceService.logDumpEvent(dataId, group, tenant, null, lastModified, handleIp,
@@ -261,7 +261,7 @@ class DumpAllProcessor implements TaskProcessor {
                     }
 
                     boolean result = ConfigService.dump(cf.getDataId(), cf.getGroup(), cf.getTenant(), cf.getContent(),
-                        cf.getLastModified());
+                        cf.getLastModified(), cf.getType());
 
                     final String content = cf.getContent();
                     final String md5 = MD5.getInstance().getMD5String(content);

test/src/test/java/com/alibaba/nacos/test/config/ConfigLongPollReturnChanges_ITCase.java

@@ -0,0 +1,151 @@
+/*
+ * 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.test.config;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.ConfigChangeEvent;
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.PropertyChangeType;
+import com.alibaba.nacos.api.exception.NacosException;
+import com.alibaba.nacos.client.config.listener.impl.AbstractConfigChangeListener;
+import com.alibaba.nacos.config.server.Config;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = Config.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+public class ConfigLongPollReturnChanges_ITCase {
+    @LocalServerPort
+    private int port;
+
+    private ConfigService configService;
+
+    @Before
+    public void init() throws NacosException {
+        Properties properties = new Properties();
+        properties.put(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:" + port);
+        properties.put(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT, "20000");
+        properties.put(PropertyKeyConst.CONFIG_RETRY_TIME, "3000");
+        properties.put(PropertyKeyConst.MAX_RETRY, "5");
+        configService = NacosFactory.createConfigService(properties);
+    }
+
+    @Test
+    public void testAdd() throws InterruptedException, NacosException {
+        CountDownLatch latch = new CountDownLatch(1);
+
+        final String dataId = "test" + System.currentTimeMillis();
+        final String group = "DEFAULT_GROUP";
+        final String content = "config data";
+
+        configService.addListener(dataId, group, new AbstractConfigChangeListener() {
+            @Override
+            public void receiveConfigChange(ConfigChangeEvent event) {
+                ConfigChangeItem cci = event.getChangeItem("content");
+                Assert.assertEquals(null, cci.getOldValue());
+                Assert.assertEquals(content, cci.getNewValue());
+                Assert.assertEquals(PropertyChangeType.ADDED, cci.getType());
+                System.out.println(cci);
+                latch.countDown();
+            }
+
+        });
+        configService.publishConfig(dataId, group, content);
+
+        latch.await();
+    }
+
+    @Test
+    public void testModify() throws InterruptedException, NacosException {
+        CountDownLatch latch = new CountDownLatch(1);
+
+        final String dataId = "test" + System.currentTimeMillis();
+        final String group = "DEFAULT_GROUP";
+        final String oldData = "old data";
+        final String newData = "new data";
+
+        configService.publishConfig(dataId, group, oldData);
+
+        // query config immediately may return null
+        String config = null;
+        do {
+            TimeUnit.SECONDS.sleep(1);
+            config = configService.getConfig(dataId, group, 50);
+        } while(null == config);
+
+        configService.addListener(dataId, group, new AbstractConfigChangeListener() {
+            @Override
+            public void receiveConfigChange(ConfigChangeEvent event) {
+                ConfigChangeItem cci = event.getChangeItem("content");
+                Assert.assertEquals(oldData, cci.getOldValue());
+                Assert.assertEquals(newData, cci.getNewValue());
+                Assert.assertEquals(PropertyChangeType.MODIFIED, cci.getType());
+                System.out.println(cci);
+                latch.countDown();
+            }
+
+        });
+        configService.publishConfig(dataId, group, newData);
+
+        latch.await();
+    }
+
+    @Test
+    public void testDelete() throws InterruptedException, NacosException {
+        CountDownLatch latch = new CountDownLatch(1);
+
+        final String dataId = "test" + System.currentTimeMillis();
+        final String group = "DEFAULT_GROUP";
+        final String oldData = "old data";
+
+        configService.publishConfig(dataId, group, oldData);
+
+        // query config immediately may return null
+        String config = null;
+        do {
+            TimeUnit.SECONDS.sleep(1);
+            config = configService.getConfig(dataId, group, 50);
+        } while(null == config);
+
+        configService.addListener(dataId, group, new AbstractConfigChangeListener() {
+            @Override
+            public void receiveConfigChange(ConfigChangeEvent event) {
+                ConfigChangeItem cci = event.getChangeItem("content");
+                Assert.assertEquals(oldData, cci.getOldValue());
+                Assert.assertEquals(null, cci.getNewValue());
+                Assert.assertEquals(PropertyChangeType.DELETED, cci.getType());
+                System.out.println(cci);
+                latch.countDown();
+            }
+
+        });
+        configService.removeConfig(dataId, group);
+
+        latch.await();
+    }
+
+}

test/src/test/java/com/alibaba/nacos/test/config/TextChangeParser.java

@@ -0,0 +1,51 @@
+/*
+ * 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.test.config;
+
+import com.alibaba.nacos.api.config.ConfigChangeItem;
+import com.alibaba.nacos.api.config.PropertyChangeType;
+import com.alibaba.nacos.api.config.listener.ConfigChangeParser;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class TextChangeParser implements ConfigChangeParser {
+    @Override
+    public boolean isResponsibleFor(String type) {
+        return (null == type || "text".equalsIgnoreCase(type));
+    }
+
+    @Override
+    public Map<String, ConfigChangeItem> doParse(String oldContent, String newContent, String type) throws IOException {
+        Map<String, ConfigChangeItem> map = new HashMap<>(4);
+        final String key = "content";
+
+        ConfigChangeItem cci = new ConfigChangeItem(key, oldContent, newContent);
+        if (null == oldContent && null != newContent) {
+            cci.setType(PropertyChangeType.ADDED);
+        } else if (null != oldContent && null != newContent && !oldContent.equals(newContent)) {
+            cci.setType(PropertyChangeType.MODIFIED);
+        } else if (null != oldContent && null == newContent) {
+            cci.setType(PropertyChangeType.DELETED);
+        }
+        map.put(key, cci);
+
+        return map;
+    }
+}
+

test/src/test/resources/META-INF/services/com.alibaba.nacos.api.config.listener.ConfigChangeParser

@@ -0,0 +1 @@
+com.alibaba.nacos.test.config.TextChangeParser