/*
 * 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.istio.common;

import com.alibaba.nacos.istio.mcp.NacosMcpService;
import com.alibaba.nacos.istio.misc.Loggers;
import com.alibaba.nacos.istio.model.PushRequest;
import com.alibaba.nacos.istio.util.IstioExecutor;
import com.alibaba.nacos.istio.xds.NacosXdsService;
import com.alibaba.nacos.sys.utils.ApplicationUtils;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * EventProcessor.
 *
 * @author special.fy
 */
@Component
public class EventProcessor implements ApplicationListener<ContextRefreshedEvent> {
    
    private static final int MAX_WAIT_EVENT_TIME = 100;
    
    private NacosMcpService nacosMcpService;
    
    private NacosXdsService nacosXdsService;
    
    private NacosResourceManager resourceManager;
    
    private final BlockingQueue<PushRequest> requests;
    
    public EventProcessor() {
        requests = new ArrayBlockingQueue<>(20);
    }
    
    /**
     * notify.
     *
     * @param pushRequest push request
     */
    public void notify(PushRequest pushRequest) {
        try {
            requests.put(pushRequest);
        } catch (InterruptedException e) {
            Loggers.MAIN.warn("There are too many events, this event {} will be ignored.", pushRequest.getReason());
            // set the interrupted flag
            Thread.currentThread().interrupt();
        }
    }
    
    private void handleEvents() {
        Consumer handleEvents = new Consumer("handle events");
        handleEvents.setDaemon(true);
        handleEvents.start();
    }
    
    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        if (contextRefreshedEvent.getApplicationContext().getParent() == null) {
            checkDependenceReady();
            handleEvents();
        }
    }
    
    private class Consumer extends Thread {
        
        Consumer(String name) {
            setName(name);
        }
        
        @Override
        @SuppressWarnings("InfiniteLoopStatement")
        public void run() {
            Future<Void> task = null;
            boolean hasNewEvent = false;
            PushRequest lastEvent = null;
            while (true) {
                try {
                    // Today we only care about service event,
                    // so we simply ignore event until the last task has been completed.
                    PushRequest pushRequest = requests.poll(MAX_WAIT_EVENT_TIME, TimeUnit.MILLISECONDS);
                    if (pushRequest != null) {
                        hasNewEvent = true;
                        lastEvent = pushRequest;
                    }
                    if (hasClientConnection() && needNewTask(hasNewEvent, task)) {
                        task = IstioExecutor.asyncHandleEvent(new EventHandleTask(lastEvent));
                        hasNewEvent = false;
                        lastEvent = null;
                    }
                } catch (InterruptedException e) {
                    Loggers.MAIN.warn("Thread {} is be interrupted.", getName());
                    // set the interrupted flag
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
    
    private boolean hasClientConnection() {
        return nacosMcpService.hasClientConnection() || nacosXdsService.hasClientConnection();
    }
    
    private boolean needNewTask(boolean hasNewEvent, Future<Void> task) {
        return hasNewEvent && (task == null || task.isDone());
    }
    
    private class EventHandleTask implements Callable<Void> {
        
        private final PushRequest pushRequest;
        
        EventHandleTask(PushRequest pushRequest) {
            this.pushRequest = pushRequest;
        }
        
        @Override
        public Void call() throws Exception {
            ResourceSnapshot snapshot = resourceManager.createResourceSnapshot();
            pushRequest.setResourceSnapshot(snapshot);
            nacosXdsService.handleEvent(pushRequest);
            nacosXdsService.handleDeltaEvent(pushRequest);
            nacosMcpService.handleEvent(pushRequest);
            return null;
        }
    }
    
    private boolean checkDependenceReady() {
        if (null == resourceManager) {
            resourceManager = ApplicationUtils.getBean(NacosResourceManager.class);
        }
        if (null == nacosXdsService) {
            nacosXdsService = ApplicationUtils.getBean(NacosXdsService.class);
        }
        if (null == nacosMcpService) {
            nacosMcpService = ApplicationUtils.getBean(NacosMcpService.class);
        }
        return Objects.nonNull(resourceManager) && Objects.nonNull(nacosMcpService) && Objects.nonNull(nacosXdsService);
    }
}
