/*
 * 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.config.server.service.notify;

import com.alibaba.nacos.common.executor.ExecutorFactory;
import com.alibaba.nacos.common.executor.NameThreadFactory;
import com.alibaba.nacos.common.task.NacosTask;
import com.alibaba.nacos.config.server.utils.LogUtil;
import com.alibaba.nacos.core.cluster.Member;
import com.alibaba.nacos.core.cluster.ServerMemberManager;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Notify Single server.
 *
 * @author Nacos
 */
public class NotifySingleService {
    
    static class NotifyTaskProcessorWrapper extends NotifyTaskProcessor {
        
        public NotifyTaskProcessorWrapper() {
            /*
             *  serverListManager is useless here
             */
            super(null);
        }
        
        @Override
        public boolean process(NacosTask task) {
            NotifySingleTask notifyTask = (NotifySingleTask) task;
            return notifyToDump(notifyTask.getDataId(), notifyTask.getGroup(), notifyTask.getTenant(),
                    notifyTask.getLastModified(), notifyTask.target);
        }
    }
    
    static class NotifySingleTask extends NotifyTask implements Runnable {
        
        private static final NotifyTaskProcessorWrapper PROCESSOR = new NotifyTaskProcessorWrapper();
        
        private final Executor executor;
        
        private final String target;
        
        private boolean isSuccess = false;
        
        public NotifySingleTask(String dataId, String group, String tenant, long lastModified, String target,
                Executor executor) {
            super(dataId, group, tenant, lastModified);
            this.target = target;
            this.executor = executor;
        }
        
        @Override
        public void run() {
            try {
                this.isSuccess = PROCESSOR.process(this);
            } catch (Exception e) { // never goes here, but in case (never interrupts this notification thread)
                this.isSuccess = false;
                LogUtil.NOTIFY_LOG
                        .error("[notify-exception] target:{} dataid:{} group:{} ts:{}", target, getDataId(), getGroup(),
                                getLastModified());
                LogUtil.NOTIFY_LOG.debug("[notify-exception] target:{} dataid:{} group:{} ts:{}",
                        new Object[] {target, getDataId(), getGroup(), getLastModified()}, e);
            }
            
            if (!this.isSuccess) {
                LogUtil.NOTIFY_LOG
                        .error("[notify-retry] target:{} dataid:{} group:{} ts:{}", target, getDataId(), getGroup(),
                                getLastModified());
                try {
                    ((ScheduledThreadPoolExecutor) executor).schedule(this, 500L, TimeUnit.MILLISECONDS);
                } catch (Exception e) { // The notification failed, but at the same time, the node was offline
                    LOGGER.warn("[notify-thread-pool] cluster remove node {}, current thread was tear down.", target,
                            e);
                }
            }
        }
    }
    
    @Autowired
    public NotifySingleService(ServerMemberManager memberManager) {
        this.memberManager = memberManager;
        setupNotifyExecutors();
    }
    
    /**
     * When the system is started or when the cluster is expanded or offline: single-threaded setupNotifyExecutors
     * executors use ConcurrentHashMap to ensure visibility.
     */
    private void setupNotifyExecutors() {
        Collection<Member> clusterIps = memberManager.allMembers();
        
        for (Member member : clusterIps) {
            
            final String address = member.getAddress();
            
            /*
             * Fixed number of threads, unbounded queue
             * (based on assumption: thread pool throughput is good,
             * there will be no continuous task accumulation,
             * there is occasional instantaneous pressure)
             */
            Executor executor = ExecutorFactory.newSingleScheduledExecutorService(
                    new NameThreadFactory("com.alibaba.nacos.config.NotifySingleServiceThread-" + address));
            
            if (null == executors.putIfAbsent(address, executor)) {
                LOGGER.warn("[notify-thread-pool] setup thread target ip {} ok.", address);
            }
        }
        
        for (Map.Entry<String, Executor> entry : executors.entrySet()) {
            String target = entry.getKey();
            
            // The cluster node goes offline
            if (!clusterIps.contains(target)) {
                ThreadPoolExecutor executor = (ThreadPoolExecutor) entry.getValue();
                executor.shutdown();
                executors.remove(target);
                LOGGER.warn("[notify-thread-pool] tear down thread target ip {} ok.", target);
            }
        }
        
    }
    
    private static final Logger LOGGER = LogUtil.FATAL_LOG;
    
    private ServerMemberManager memberManager;
    
    private ConcurrentHashMap<String, Executor> executors = new ConcurrentHashMap<String, Executor>();
    
    public ConcurrentHashMap<String, Executor> getExecutors() {
        return executors;
    }
}

@@ -703,487 +703,103 @@
 mon.
-executor.ExecutorFactory;%0Aimport com.alibaba.nacos.common.executor.NameThreadFactory;%0Aimport com.alibaba.nacos.common.task.NacosTask;%0Aimport com.alibaba.nacos.config.server.utils.LogUtil;%0Aimport com.alibaba.nacos.core.cluster.Member;%0Aimport com.alibaba.nacos.core.cluster.ServerMemberManager;%0Aimport org.slf4j.Logger;%0Aimport org.springframework.beans.factory.annotation.Autowired;%0A%0Aimport java.util.Collection;%0Aimport java.util.Map;%0Aimport java.util.concurrent.ConcurrentHashMap;
+task.NacosTask;%0Aimport com.alibaba.nacos.config.server.utils.LogUtil;%0Aimport org.slf4j.Logger;%0A
 %0Aimp
@@ -894,56 +894,8 @@
 or;%0A
-import java.util.concurrent.ThreadPoolExecutor;%0A
 impo
@@ -1013,24 +1013,84 @@
 leService %7B%0A
+    private static final Logger LOGGER = LogUtil.FATAL_LOG;%0A
     %0A    sta
@@ -2909,22 +2909,8 @@
     
-new Object%5B%5D %7B
 targ
@@ -2955,17 +2955,16 @@
 dified()
-%7D
 , e);%0A  
@@ -3668,2139 +3668,6 @@
   %7D%0A
-    %0A    @Autowired%0A    public NotifySingleService(ServerMemberManager memberManager) %7B%0A        this.memberManager = memberManager;%0A        setupNotifyExecutors();%0A    %7D%0A    %0A    /**%0A     * When the system is started or when the cluster is expanded or offline: single-threaded setupNotifyExecutors%0A     * executors use ConcurrentHashMap to ensure visibility.%0A     */%0A    private void setupNotifyExecutors() %7B%0A        Collection%3CMember%3E clusterIps = memberManager.allMembers();%0A        %0A        for (Member member : clusterIps) %7B%0A            %0A            final String address = member.getAddress();%0A            %0A            /*%0A             * Fixed number of threads, unbounded queue%0A             * (based on assumption: thread pool throughput is good,%0A             * there will be no continuous task accumulation,%0A             * there is occasional instantaneous pressure)%0A             */%0A            Executor executor = ExecutorFactory.newSingleScheduledExecutorService(%0A                    new NameThreadFactory(%22com.alibaba.nacos.config.NotifySingleServiceThread-%22 + address));%0A            %0A            if (null == executors.putIfAbsent(address, executor)) %7B%0A                LOGGER.warn(%22%5Bnotify-thread-pool%5D setup thread target ip %7B%7D ok.%22, address);%0A            %7D%0A        %7D%0A        %0A        for (Map.Entry%3CString, Executor%3E entry : executors.entrySet()) %7B%0A            String target = entry.getKey();%0A            %0A            // The cluster node goes offline%0A            if (!clusterIps.contains(target)) %7B%0A                ThreadPoolExecutor executor = (ThreadPoolExecutor) entry.getValue();%0A                executor.shutdown();%0A                executors.remove(target);%0A                LOGGER.warn(%22%5Bnotify-thread-pool%5D tear down thread target ip %7B%7D ok.%22, target);%0A            %7D%0A        %7D%0A        %0A    %7D%0A    %0A    private static final Logger LOGGER = LogUtil.FATAL_LOG;%0A    %0A    private ServerMemberManager memberManager;%0A    %0A    private ConcurrentHashMap%3CString, Executor%3E executors = new ConcurrentHashMap%3CString, Executor%3E();%0A    %0A    public ConcurrentHashMap%3CString, Executor%3E getExecutors() %7B%0A        return executors;%0A    %7D%0A
 %7D%0A

Change the invalid compare in NotifySingleService (#5587)