/*******************************************************************************
 * Copyright 2016-2017 ZTE, Inc. and others.
 * 
 * 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 org.onap.msb.apiroute.wrapper.queue;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.http.HttpEntity;
import org.onap.msb.apiroute.SyncDataManager;
import org.onap.msb.apiroute.wrapper.MicroServiceWrapper;
import org.onap.msb.apiroute.wrapper.util.CommonUtil;
import org.onap.msb.apiroute.wrapper.util.ServiceFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;


public class ServiceListConsumer implements Runnable {

	private static final Logger LOGGER = LoggerFactory
			.getLogger(ServiceListConsumer.class);

	private boolean isRunning = true;

	private int index;


	public ServiceListConsumer() {
		this.index = 0;
	}

	public void run() {
		LOGGER.info("run ServiceList Consumer Thread [" + index + "]");

		while (isRunning) {
			try {
				// 取最新一条记录
				ServiceData<HttpEntity> serviceData = QueueManager
						.getInstance().takeFromServiceListQueue(index);
				 LOGGER.debug("ServiceList Consumer Thread [" + index +
				 "]  take out serviceData from Queue successfully");

				HttpEntity newValues = serviceData.getData();

				Set<String> newServiceNameList = filterServiceList(newValues);

				if (ServiceListCache.getLatestServiceNamelist().size() == 0) {
					boolean initSuccess=initServiceList(newServiceNameList);
					if(initSuccess){
					  ServiceListCache.setLatestServiceNamelist(newServiceNameList);
					}
				} else {
					updateServiceList(newServiceNameList);
					ServiceListCache.setLatestServiceNamelist(newServiceNameList);
				}

				
			} catch (Exception e) {
				LOGGER.error(
						"ServiceListConsumer throw  Exception: ", e);
			}
		}
	}

	private void startWatchService(String serviceName) {
		// start to Watch service nodes

		SyncDataManager.startWatchService(serviceName);
	}

	private void updateServiceList(Set<String> newServiceNameList) {
		Set<String> registerServiceNameList = CommonUtil.getDiffrent(
		    ServiceListCache.getLatestServiceNamelist(), newServiceNameList);

		if (registerServiceNameList.size() > 0) {
			LOGGER.info("***need to start Watch Service num from consul :"
					+ registerServiceNameList.size());

			for (String serviceName : registerServiceNameList) {
				startWatchService(serviceName);
			}
		}
	}

	private boolean initServiceList(Set<String> newServiceNameList) {
		LOGGER.info("***start to initialize  service List when System startup ***");

		Set<String> dbServiceNameList = MicroServiceWrapper
				.getInstance().getAllMicroServiceKey();
		
		if(dbServiceNameList==null){		 
		  LOGGER.error("init ServiceList from redis fail ");
		  return false;
		}
		
		
    		// 对比删除redis脏数据
    		Set<String> delServiceNameList = CommonUtil.getDiffrent(
    				newServiceNameList, dbServiceNameList);
    
    		LOGGER.info("***need to delete Service num from redis :"
    				+ delServiceNameList.size());
    		for (String serviceName : delServiceNameList) {
    			try {
    				MicroServiceWrapper.getInstance()
    						.deleteMicroService4AllVersion(serviceName);
    				LOGGER.info("delete MicroService success from initialize:[serviceName]"
    						+ serviceName);
    
    			} catch (Exception e) {
    				LOGGER.error(
    						"initialize serviceList :Delete MicroServiceInfo serviceName:"
    								+ serviceName + " FAIL : ", e);
    			}
    		}
    
    		// 启动同步开启监听全部服务列表
    		LOGGER.info("***need to start Watch Service num from initialize :"
    				+ newServiceNameList.size());
    
    		for (String serviceName : newServiceNameList) {
    			startWatchService(serviceName);
    		}
		
    		return true;

	}

	/*private ImmutableSet<String> filterServiceList(
			final Map<String, List<String>> serviceList) {
		if (serviceList == null || serviceList.isEmpty()) {
			return ImmutableSet.of();
		}

		final ImmutableSet.Builder<String> builder = ImmutableSet.builder();

		for (Map.Entry<String, List<String>> entry : serviceList.entrySet()) {

			String key = entry.getKey();
			if (key != null && !"consul".equals(key)) {

				List<String> value = entry.getValue();
				if (ServiceFilter.getInstance().isFilterService(value)) {
					builder.add(key);
				}
			}
		}

		LOGGER.info("consul all service num:" + serviceList.size());
		LOGGER.info("consul filter service num:" + builder.build().size());

		return builder.build();
	}
*/
	private Set<String> filterServiceList(final HttpEntity serviceList) {

		if (serviceList == null || serviceList.getContentLength() == 0) {
			return new HashSet<String>();
		}

		final Set<String> builder = new HashSet<String>();

		JsonFactory f = new JsonFactory();
		JsonParser jp = null;
		List<String> tagList = null;
		int inputServiceNum = 0;
		try {
			jp = f.createParser(serviceList.getContent());
			jp.nextToken();
			while (jp.nextToken() != JsonToken.END_OBJECT) {
				String serviceName = jp.getCurrentName();
				inputServiceNum++;
				jp.nextToken();
				tagList = new ArrayList<>();
				while (jp.nextToken() != JsonToken.END_ARRAY) {
					tagList.add(jp.getText());
				}
				
				if (serviceName != null && !"consul".equals(serviceName)) {
					if (ServiceFilter.getInstance().isFilterService(tagList)) {
						builder.add(serviceName);
					}
				}
			}
		} catch (IOException e) {
			LOGGER.warn("parse service list error",e);
			return new HashSet<String>();
		} finally {
			try {
				jp.close();
			} catch (IOException e) {
				LOGGER.warn("parse service list error",e);
				return new HashSet<String>();
			}
		}
		
		int latestServiceNum=ServiceListCache.getLatestServiceNamelist().size();
//		if(latestServiceNum!=builder.size()){
		  LOGGER.info("[consul] all service num:" + inputServiceNum+ ", filter service num: new——" + builder.size()+"  old——"+latestServiceNum);
//		}

		return builder;
	}

}