/*
 * 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.slots.block.flow.param;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;

import com.alibaba.csp.sentinel.cluster.client.ClusterTokenClient;
import com.alibaba.csp.sentinel.cluster.client.TokenClientProvider;
import com.alibaba.csp.sentinel.cluster.TokenResult;
import com.alibaba.csp.sentinel.cluster.TokenResultStatus;
import com.alibaba.csp.sentinel.log.RecordLog;
import com.alibaba.csp.sentinel.slotchain.ResourceWrapper;
import com.alibaba.csp.sentinel.slots.block.RuleConstant;

/**
 * Rule checker for parameter flow control.
 *
 * @author Eric Zhao
 * @since 0.2.0
 */
final class ParamFlowChecker {

    static boolean passCheck(ResourceWrapper resourceWrapper, /*@Valid*/ ParamFlowRule rule, /*@Valid*/ int count,
                             Object... args) {
        if (args == null) {
            return true;
        }

        int paramIdx = rule.getParamIdx();
        if (args.length <= paramIdx) {
            return true;
        }

        // Get parameter value. If value is null, then pass.
        Object value = args[paramIdx];
        if (value == null) {
            return true;
        }

        if (rule.isClusterMode()) {
            return passClusterCheck(resourceWrapper, rule, count, value);
        }

        return passLocalCheck(resourceWrapper, rule, count, value);
    }

    private static boolean passLocalCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count,
                                          Object value) {
        try {
            if (Collection.class.isAssignableFrom(value.getClass())) {
                for (Object param : ((Collection)value)) {
                    if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {
                        return false;
                    }
                }
            } else if (value.getClass().isArray()) {
                int length = Array.getLength(value);
                for (int i = 0; i < length; i++) {
                    Object param = Array.get(value, i);
                    if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {
                        return false;
                    }
                }
            } else {
                return passSingleValueCheck(resourceWrapper, rule, count, value);
            }
        } catch (Throwable e) {
            RecordLog.warn("[ParamFlowChecker] Unexpected error", e);
        }

        return true;
    }

    static boolean passSingleValueCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count, Object value) {
        Set<Object> exclusionItems = rule.getParsedHotItems().keySet();
        if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {
            double curCount = getHotParameters(resourceWrapper).getPassParamQps(rule.getParamIdx(), value);

            if (exclusionItems.contains(value)) {
                // Pass check for exclusion items.
                int itemQps = rule.getParsedHotItems().get(value);
                return curCount + count <= itemQps;
            } else if (curCount + count > rule.getCount()) {
                if ((curCount - rule.getCount()) < 1 && (curCount - rule.getCount()) > 0) {
                    return true;
                }
                return false;
            }
        }

        return true;
    }

    private static ParameterMetric getHotParameters(ResourceWrapper resourceWrapper) {
        // Should not be null.
        return ParamFlowSlot.getParamMetric(resourceWrapper);
    }

    @SuppressWarnings("unchecked")
    private static Collection<Object> toCollection(Object value) {
        if (value instanceof Collection) {
            return (Collection<Object>)value;
        } else if (value.getClass().isArray()) {
            List<Object> params = new ArrayList<Object>();
            int length = Array.getLength(value);
            for (int i = 0; i < length; i++) {
                Object param = Array.get(value, i);
                params.add(param);
            }
            return params;
        } else {
            return Collections.singletonList(value);
        }
    }

    private static boolean passClusterCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count,
                                            Object value) {
        try {
            ClusterTokenClient client = TokenClientProvider.getClient();
            if (client == null) {
                return true;
            }
            Collection<Object> params = toCollection(value);

            TokenResult result = client.requestParamToken(rule.getClusterConfig().getFlowId(), count, params);
            switch (result.getStatus()) {
                case TokenResultStatus.OK:
                    return true;
                case TokenResultStatus.BLOCKED:
                    return false;
                default:
                    return fallbackToLocalOrPass(resourceWrapper, rule, count, params);
            }
        } catch (Throwable ex) {
            RecordLog.warn("[ParamFlowChecker] Request cluster token for parameter unexpected failed", ex);
            return fallbackToLocalOrPass(resourceWrapper, rule, count, value);
        }
    }

    private static boolean fallbackToLocalOrPass(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count,
                                                 Object value) {
        if (rule.getClusterConfig().isFallbackToLocalWhenFail()) {
            return passLocalCheck(resourceWrapper, rule, count, value);
        } else {
            // The rule won't be activated, just pass.
            return true;
        }
    }

    private ParamFlowChecker() {}
}

@@ -876,33 +876,81 @@
 ter.
-client.C
+ClusterStateManager;%0Aimport com.alibaba.csp.sentinel.c
 luster
+.
 Token
-Client
+Service
 ;%0Aim
@@ -1127,16 +1127,99 @@
 Status;%0A
+import com.alibaba.csp.sentinel.cluster.server.EmbeddedClusterTokenServerProvider;%0A
 import c
@@ -5177,64 +5177,117 @@
    C
-lusterTokenClient client = TokenClientProvider.getClient
+ollection%3CObject%3E params = toCollection(value);%0A%0A            TokenService clusterService = pickClusterService
 ();%0A
@@ -5304,20 +5304,28 @@
   if (cl
-ient
+usterService
  == null
@@ -5348,95 +5348,182 @@
     
-return true;%0A            %7D%0A            Collection%3CObject%3E params = toCollection(value);
+// No available cluster client or server, fallback to local or pass in need.%0A                return fallbackToLocalOrPass(resourceWrapper, rule, count, params);%0A            %7D
 %0A%0A  
@@ -5555,20 +5555,28 @@
 ult = cl
-ient
+usterService
 .request
@@ -6626,32 +6626,346 @@
        %7D%0A    %7D%0A%0A
+    private static TokenService pickClusterService() %7B%0A        if (ClusterStateManager.isClient()) %7B%0A            return TokenClientProvider.getClient();%0A        %7D%0A        if (ClusterStateManager.isServer()) %7B%0A            return EmbeddedClusterTokenServerProvider.getServer();%0A        %7D%0A        return null;%0A    %7D%0A%0A
     private Para

Add cluster embedded mode handling logic for parameter flow checker