Add Nacos datasource integration

Signed-off-by: Eric Zhao <sczyh16@gmail.com>

sentinel-core/src/main/java/com/alibaba/csp/sentinel/property/DynamicSentinelProperty.java

@@ -59,12 +59,12 @@ public class DynamicSentinelProperty<T> implements SentinelProperty<T> {
 
     }
 
-    public boolean isEqual(T oldValue, T newValue) {
+    private boolean isEqual(T oldValue, T newValue) {
         if (oldValue == null && newValue == null) {
             return true;
         }
 
-        if (oldValue == null && newValue != null) {
+        if (oldValue == null) {
             return false;
         }
 

sentinel-extension/pom.xml

@@ -13,6 +13,7 @@
 
     <modules>
         <module>sentinel-datasource-extension</module>
+        <module>sentinel-datasource-nacos</module>
     </modules>
 
 </project>

sentinel-extension/sentinel-datasource-extension/pom.xml

@@ -17,10 +17,7 @@
             <groupId>com.alibaba.csp</groupId>
             <artifactId>sentinel-core</artifactId>
         </dependency>
-        <dependency>
-            <groupId>com.alibaba</groupId>
-            <artifactId>fastjson</artifactId>
-        </dependency>
+
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

sentinel-extension/sentinel-datasource-extension/src/main/java/com/alibaba/csp/sentinel/datasource/AbstractDataSource.java

@@ -20,8 +20,8 @@ import com.alibaba.csp.sentinel.property.SentinelProperty;
 
 public abstract class AbstractDataSource<S, T> implements DataSource<S, T> {
 
-    protected ConfigParser<S, T> parser;
-    protected SentinelProperty<T> property;
+    protected final ConfigParser<S, T> parser;
+    protected final SentinelProperty<T> property;
 
     public AbstractDataSource(ConfigParser<S, T> parser) {
         if (parser == null) {

sentinel-extension/sentinel-datasource-nacos/pom.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>sentinel-extension</artifactId>
+        <groupId>com.alibaba.csp</groupId>
+        <version>0.1.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-datasource-nacos</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <nacos.version>0.1.0</nacos.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-datasource-extension</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.nacos</groupId>
+            <artifactId>nacos-client</artifactId>
+            <version>${nacos.version}</version>
+        </dependency>
+    </dependencies>
+</project>

sentinel-extension/sentinel-datasource-nacos/src/main/java/com/alibaba/csp/sentinel/datasource/nacos/NacosDataSource.java

@@ -0,0 +1,134 @@
+/*
+ * 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.csp.sentinel.datasource.nacos;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import com.alibaba.csp.sentinel.concurrent.NamedThreadFactory;
+import com.alibaba.csp.sentinel.datasource.AbstractDataSource;
+import com.alibaba.csp.sentinel.datasource.ConfigParser;
+import com.alibaba.csp.sentinel.datasource.DataSource;
+import com.alibaba.csp.sentinel.log.RecordLog;
+import com.alibaba.csp.sentinel.util.StringUtil;
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.Listener;
+
+/**
+ * A {@link DataSource} with Nacos backend. When the data in Nacos backend has been modified,
+ * Nacos will automatically push the new value so that the dynamic configuration can be real-time.
+ *
+ * @author Eric Zhao
+ */
+public class NacosDataSource<T> extends AbstractDataSource<String, T> {
+
+    private static final int DEFAULT_TIMEOUT = 3000;
+
+    /**
+     * Single-thread pool. Once the thread pool is blocked, we throw up the old task.
+     */
+    private final ExecutorService pool = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+        new ArrayBlockingQueue<Runnable>(1), new NamedThreadFactory("sentinel-nacos-ds-update"),
+        new ThreadPoolExecutor.DiscardOldestPolicy());
+
+    private final Listener configListener;
+    private final String groupId;
+    private final String dataId;
+
+    /**
+     * Note: The Nacos config might be null if its initialization failed.
+     */
+    private ConfigService configService = null;
+
+    /**
+     * Constructs an DataSource with Nacos backend.
+     *
+     * @param serverAddr server address of Nacos, cannot be empty
+     * @param groupId    group ID, cannot be empty
+     * @param dataId     data ID, cannot be empty
+     * @param parser     customized data parser, cannot be empty
+     */
+    public NacosDataSource(final String serverAddr, final String groupId, final String dataId,
+                           ConfigParser<String, T> parser) {
+        super(parser);
+        if (StringUtil.isBlank(serverAddr) || StringUtil.isBlank(groupId) || StringUtil.isBlank(dataId)) {
+            throw new IllegalArgumentException(String.format("Bad argument: serverAddr=[%s], groupId=[%s], dataId=[%s]",
+                serverAddr, groupId, dataId));
+        }
+        this.groupId = groupId;
+        this.dataId = dataId;
+        this.configListener = new Listener() {
+            @Override
+            public Executor getExecutor() {
+                return pool;
+            }
+
+            @Override
+            public void receiveConfigInfo(final String configInfo) {
+                RecordLog.info(String.format("[NacosDataSource] New property value received for (%s, %s, %s): %s",
+                    serverAddr, dataId, groupId, configInfo));
+                T newValue = NacosDataSource.this.parser.parse(configInfo);
+                // Update the new value to the property.
+                getProperty().updateValue(newValue);
+            }
+        };
+        initNacosListener(serverAddr);
+        loadInitialConfig();
+    }
+
+    private void loadInitialConfig() {
+        try {
+            T newValue = loadConfig();
+            if (newValue == null) {
+                RecordLog.info("[NacosDataSource] WARN: initial config is null, you may have to check your data source");
+            }
+            getProperty().updateValue(newValue);
+        } catch (Exception ex) {
+            RecordLog.info("[NacosDataSource] Error when loading initial config", ex);
+        }
+    }
+
+    private void initNacosListener(String serverAddr) {
+        try {
+            this.configService = NacosFactory.createConfigService(serverAddr);
+            // Add config listener.
+            configService.addListener(dataId, groupId, configListener);
+        } catch (Exception e) {
+            RecordLog.info("[NacosDataSource] Error occurred when initializing Nacos data source", e);
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public String readSource() throws Exception {
+        if (configService == null) {
+            throw new IllegalStateException("Nacos config service has not been initialized or error occurred");
+        }
+        return configService.getConfig(dataId, groupId, DEFAULT_TIMEOUT);
+    }
+
+    @Override
+    public void close() {
+        if (configService != null) {
+            configService.removeListener(dataId, groupId, configListener);
+        }
+        pool.shutdownNow();
+    }
+}