Support registering writable data-source for GatewayFlowRule and customized ApiDefinition (#1057)

sentinel-adapter/sentinel-api-gateway-adapter-common/src/main/java/com/alibaba/csp/sentinel/adapter/gateway/common/command/UpdateGatewayApiDefinitionGroupCommandHandler.java

@@ -15,10 +15,6 @@
  */
 package com.alibaba.csp.sentinel.adapter.gateway.common.command;
 
-import java.net.URLDecoder;
-import java.util.HashSet;
-import java.util.Set;
-
 import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiDefinition;
 import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiPathPredicateItem;
 import com.alibaba.csp.sentinel.adapter.gateway.common.api.ApiPredicateItem;
@@ -27,12 +23,17 @@ import com.alibaba.csp.sentinel.command.CommandHandler;
 import com.alibaba.csp.sentinel.command.CommandRequest;
 import com.alibaba.csp.sentinel.command.CommandResponse;
 import com.alibaba.csp.sentinel.command.annotation.CommandMapping;
+import com.alibaba.csp.sentinel.datasource.WritableDataSource;
 import com.alibaba.csp.sentinel.log.RecordLog;
 import com.alibaba.csp.sentinel.util.StringUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 
+import java.net.URLDecoder;
+import java.util.HashSet;
+import java.util.Set;
+
 /**
  * @author Eric Zhao
  * @since 1.6.0
@@ -40,6 +41,8 @@ import com.alibaba.fastjson.JSONObject;
 @CommandMapping(name = "gateway/updateApiDefinitions", desc = "")
 public class UpdateGatewayApiDefinitionGroupCommandHandler implements CommandHandler<String> {
 
+    private static WritableDataSource<Set<ApiDefinition>> apiDefinitionWds = null;
+
     @Override
     public CommandResponse<String> handle(CommandRequest request) {
         String data = request.getParam("data");
@@ -59,11 +62,14 @@ public class UpdateGatewayApiDefinitionGroupCommandHandler implements CommandHan
 
         Set<ApiDefinition> apiDefinitions = parseJson(data);
         GatewayApiDefinitionManager.loadApiDefinitions(apiDefinitions);
-
+        if (!writeToDataSource(apiDefinitionWds, apiDefinitions)) {
+            result = WRITE_DS_FAILURE_MSG;
+        }
         return CommandResponse.ofSuccess(result);
     }
 
     private static final String SUCCESS_MSG = "success";
+    private static final String WRITE_DS_FAILURE_MSG = "partial success (write data source failed)";
 
     /**
      * Parse json data to set of {@link ApiDefinition}.
@@ -88,4 +94,32 @@ public class UpdateGatewayApiDefinitionGroupCommandHandler implements CommandHan
 
         return apiDefinitions;
     }
+
+    /**
+     * Write target value to given data source.
+     *
+     * @param dataSource writable data source
+     * @param value target value to save
+     * @param <T> value type
+     * @return true if write successful or data source is empty; false if error occurs
+     */
+    private <T> boolean writeToDataSource(WritableDataSource<T> dataSource, T value) {
+        if (dataSource != null) {
+            try {
+                dataSource.write(value);
+            } catch (Exception e) {
+                RecordLog.warn("Write data source failed", e);
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public synchronized static WritableDataSource<Set<ApiDefinition>> getWritableDataSource() {
+        return apiDefinitionWds;
+    }
+
+    public synchronized static void setWritableDataSource(WritableDataSource<Set<ApiDefinition>> apiDefinitionWds) {
+        UpdateGatewayApiDefinitionGroupCommandHandler.apiDefinitionWds = apiDefinitionWds;
+    }
 }

sentinel-adapter/sentinel-api-gateway-adapter-common/src/main/java/com/alibaba/csp/sentinel/adapter/gateway/common/command/UpdateGatewayRuleCommandHandler.java

@@ -15,19 +15,20 @@
  */
 package com.alibaba.csp.sentinel.adapter.gateway.common.command;
 
-import java.net.URLDecoder;
-import java.util.HashSet;
-import java.util.List;
-
 import com.alibaba.csp.sentinel.adapter.gateway.common.rule.GatewayFlowRule;
 import com.alibaba.csp.sentinel.adapter.gateway.common.rule.GatewayRuleManager;
 import com.alibaba.csp.sentinel.command.CommandHandler;
 import com.alibaba.csp.sentinel.command.CommandRequest;
 import com.alibaba.csp.sentinel.command.CommandResponse;
 import com.alibaba.csp.sentinel.command.annotation.CommandMapping;
+import com.alibaba.csp.sentinel.datasource.WritableDataSource;
 import com.alibaba.csp.sentinel.log.RecordLog;
 import com.alibaba.csp.sentinel.util.StringUtil;
-import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+
+import java.net.URLDecoder;
+import java.util.Set;
 
 /**
  * @author Eric Zhao
@@ -35,6 +36,7 @@ import com.alibaba.fastjson.JSONArray;
  */
 @CommandMapping(name = "gateway/updateRules", desc = "Update gateway rules")
 public class UpdateGatewayRuleCommandHandler implements CommandHandler<String> {
+    private static WritableDataSource<Set<GatewayFlowRule>> gatewayFlowWds = null;
 
     @Override
     public CommandResponse<String> handle(CommandRequest request) {
@@ -52,10 +54,43 @@ public class UpdateGatewayRuleCommandHandler implements CommandHandler<String> {
         RecordLog.info(String.format("[API Server] Receiving rule change (type: gateway rule): %s", data));
 
         String result = SUCCESS_MSG;
-        List<GatewayFlowRule> flowRules = JSONArray.parseArray(data, GatewayFlowRule.class);
+	    Set<GatewayFlowRule> flowRules = JSON.parseObject(data, new TypeReference<Set<GatewayFlowRule>>() {
-        GatewayRuleManager.loadRules(new HashSet<>(flowRules));
+	    });
+        GatewayRuleManager.loadRules(flowRules);
+        if (!writeToDataSource(gatewayFlowWds, flowRules)) {
+            result = WRITE_DS_FAILURE_MSG;
+        }
         return CommandResponse.ofSuccess(result);
     }
 
+    /**
+     * Write target value to given data source.
+     *
+     * @param dataSource writable data source
+     * @param value target value to save
+     * @param <T> value type
+     * @return true if write successful or data source is empty; false if error occurs
+     */
+    private <T> boolean writeToDataSource(WritableDataSource<T> dataSource, T value) {
+        if (dataSource != null) {
+            try {
+                dataSource.write(value);
+            } catch (Exception e) {
+                RecordLog.warn("Write data source failed", e);
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public synchronized static WritableDataSource<Set<GatewayFlowRule>> getWritableDataSource() {
+        return gatewayFlowWds;
+    }
+
+    public synchronized static void setWritableDataSource(WritableDataSource<Set<GatewayFlowRule>> gatewayFlowWds) {
+        UpdateGatewayRuleCommandHandler.gatewayFlowWds = gatewayFlowWds;
+    }
+
     private static final String SUCCESS_MSG = "success";
+    private static final String WRITE_DS_FAILURE_MSG = "partial success (write data source failed)";
 }