/*
 *
 * Copyright 1999-2021 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.core.remote;

import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.request.ClientDetectionRequest;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.common.remote.exception.ConnectionAlreadyClosedException;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.plugin.control.Loggers;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * nacos runtime connection ejector.
 *
 * @author shiyiyue
 */
public class NacosRuntimeConnectionEjector extends RuntimeConnectionEjector {
    
    public NacosRuntimeConnectionEjector() {
    
    }
    
    /**
     * eject connections on runtime.
     */
    public void doEject() {
        // remove out dated connection
        ejectOutdatedConnection();
        // remove overload connection
        ejectOverLimitConnection();
    }
    
    /**
     * eject the outdated connection.
     */
    private void ejectOutdatedConnection() {
        try {
            Loggers.CONNECTION.info("Connection check task start");
            Map<String, Connection> connections = connectionManager.connections;
            int totalCount = connections.size();
            int currentSdkClientCount = connectionManager.currentSdkClientCount();
            Loggers.CONNECTION.info("Long connection metrics detail ,Total count ={}, sdkCount={},clusterCount={}",
                    totalCount, currentSdkClientCount, (totalCount - currentSdkClientCount));
            Set<String> outDatedConnections = new HashSet<>();
            long now = System.currentTimeMillis();
            //outdated connections collect.
            for (Map.Entry<String, Connection> entry : connections.entrySet()) {
                Connection client = entry.getValue();
                if (now - client.getMetaInfo().getLastActiveTime() >= KEEP_ALIVE_TIME) {
                    outDatedConnections.add(client.getMetaInfo().getConnectionId());
                } else if (client.getMetaInfo().pushQueueBlockTimesLastOver(300 * 1000)) {
                    outDatedConnections.add(client.getMetaInfo().getConnectionId());
                }
            }
            // check out date connection
            Loggers.CONNECTION.info("Out dated connection ,size={}", outDatedConnections.size());
            if (CollectionUtils.isNotEmpty(outDatedConnections)) {
                Set<String> successConnections = new HashSet<>();
                final CountDownLatch latch = new CountDownLatch(outDatedConnections.size());
                for (String outDateConnectionId : outDatedConnections) {
                    try {
                        Connection connection = connectionManager.getConnection(outDateConnectionId);
                        if (connection != null) {
                            ClientDetectionRequest clientDetectionRequest = new ClientDetectionRequest();
                            connection.asyncRequest(clientDetectionRequest, new RequestCallBack() {
                                @Override
                                public Executor getExecutor() {
                                    return null;
                                }
                                
                                @Override
                                public long getTimeout() {
                                    return 5000L;
                                }
                                
                                @Override
                                public void onResponse(Response response) {
                                    latch.countDown();
                                    if (response != null && response.isSuccess()) {
                                        connection.freshActiveTime();
                                        successConnections.add(outDateConnectionId);
                                    }
                                }
                                
                                @Override
                                public void onException(Throwable e) {
                                    latch.countDown();
                                }
                            });
                            Loggers.CONNECTION.info("[{}]send connection active request ", outDateConnectionId);
                        } else {
                            latch.countDown();
                        }
                    } catch (ConnectionAlreadyClosedException e) {
                        latch.countDown();
                    } catch (Exception e) {
                        Loggers.CONNECTION.error("[{}]Error occurs when check client active detection ,error={}",
                                outDateConnectionId, e);
                        latch.countDown();
                    }
                }
                latch.await(5000L, TimeUnit.MILLISECONDS);
                Loggers.CONNECTION.info("Out dated connection check successCount={}", successConnections.size());
                for (String outDateConnectionId : outDatedConnections) {
                    if (!successConnections.contains(outDateConnectionId)) {
                        Loggers.CONNECTION.info("[{}]Unregister Out dated connection....", outDateConnectionId);
                        connectionManager.unregister(outDateConnectionId);
                    }
                }
            }
            Loggers.CONNECTION.info("Connection check task end");
        } catch (Throwable e) {
            Loggers.CONNECTION.error("Error occurs during connection check... ", e);
        }
    }
    
    /**
     * eject the over limit connection.
     */
    private void ejectOverLimitConnection() {
        // if not count set, then give up
        if (getLoadClient() > 0) {
            try {
                Loggers.CONNECTION.info("Connection overLimit check task start, loadCount={}, redirectAddress={}",
                        getLoadClient(), getRedirectAddress());
                // check count
                int currentConnectionCount = connectionManager.getCurrentConnectionCount();
                int ejectingCount = currentConnectionCount - getLoadClient();
                // if overload
                if (ejectingCount > 0) {
                    // we may modify the connection map when connection reset
                    // avoid concurrent modified exception, create new set for ids snapshot
                    Set<String> ids = new HashSet<>(connectionManager.connections.keySet());
                    for (String id : ids) {
                        if (ejectingCount > 0) {
                            // check sdk
                            Connection connection = connectionManager.getConnection(id);
                            if (connection != null && connection.getMetaInfo().isSdkSource()) {
                                if (connectionManager.loadSingle(id, redirectAddress)) {
                                    ejectingCount--;
                                }
                            }
                        } else {
                            // reach the count
                            break;
                        }
                    }
                }
                Loggers.CONNECTION.info("Connection overLimit task end, current loadCount={}, has ejected loadCont={}",
                        connectionManager.getCurrentConnectionCount(), getLoadClient() - ejectingCount);
            } catch (Throwable e) {
                Loggers.CONNECTION.error("Error occurs during connection overLimit... ", e);
            }
            // reset
            setRedirectAddress(null);
            setLoadClient(-1);
        }
    }
    
    @Override
    public String getName() {
        return "nacos";
    }
}
