package com.yeejoin.amos.kgd.message;

import com.alibaba.fastjson.JSONObject;
import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
import com.yeejoin.amos.component.influxdb.InfluxdbUtil;
import com.yeejoin.amos.kgd.message.model.MessageModel;
import net.oschina.j2cache.CacheChannel;
import net.oschina.j2cache.CacheObject;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;
import org.typroject.tyboot.component.emq.EmqxListener;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

@Component
public class MaasMessageAction extends EmqxListener{

	private InfluxDbConnection influxDbConnection;

	private InfluxdbUtil influxdbUtil;
    
	private CacheChannel cacheChannel;
	
	
	public static final int threadNum = 5;
	ExecutorService service = Executors.newFixedThreadPool(threadNum);
	private static final BlockingQueue<MessageModel> blockingQueue = new LinkedBlockingQueue<MessageModel>();

	SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
	
	public MaasMessageAction(InfluxDbConnection influxDbConnection, InfluxdbUtil influxdbUtil, CacheChannel cacheChannel) {
		this.influxDbConnection = influxDbConnection;
		this.cacheChannel = cacheChannel;
		this.influxdbUtil = influxdbUtil;
		
		for (int i = 0; i < threadNum; i++) {
			service.execute(task_runnable);
		}
	}
	
    @Override
    public void processMessage(String topic,MqttMessage mqttMessage) throws Exception {
    	MessageModel message = new MessageModel();
    	message.setPayload(mqttMessage.getPayload());
    	message.setTableName(topic.replace("/", ""));
    	blockingQueue.add(message);
    }

    Runnable task_runnable = new Runnable() {
		public void run() {
			while (true) {
				try {
					MessageModel mqttMessage = blockingQueue.take();
					String jsonStr = new String(mqttMessage.getPayload());
					JSONObject data = JSONObject.parseObject(jsonStr);
					
					String tableName = mqttMessage.getTableName();
					Map<String, String> tagsMap = new HashMap<>();
					Map<String, Object>  fieldsMap = JSONObject.toJavaObject(data, HashMap.class);

					String sql = "show field keys from " + tableName;
					try {
						List<Map<String, Object>> list = influxdbUtil.query(sql);
						Map<String, Object> fieldsTempMap = new HashMap<>();
						for (Map<String, Object> field : list) {
							if (fieldsTempMap.containsKey(field.get("fieldKey").toString())) {
								continue;
							}
							fieldsTempMap.put(field.get("fieldKey").toString(), field.get("fieldType").toString());
							if (fieldsMap.containsKey(field.get("fieldKey"))) {
								if ("integer".equals(field.get("fieldType").toString())) {
									fieldsMap.put(field.get("fieldKey").toString(),
											Integer.valueOf(fieldsMap.get(field.get("fieldKey")).toString()));
								} else if ("float".equals(field.get("fieldType").toString())) {
									fieldsMap.put(field.get("fieldKey").toString(),
											Float.valueOf(fieldsMap.get(field.get("fieldKey")).toString()));
								} else if ("double".equals(field.get("fieldType").toString())) {
									fieldsMap.put(field.get("fieldKey").toString(),
											Double.valueOf(fieldsMap.get(field.get("fieldKey")).toString()));
								} else {
									fieldsMap.put(field.get("fieldKey").toString(),
											fieldsMap.get(field.get("fieldKey")).toString());
								}
							}
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
					
					fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
					influxDbConnection.insert(tableName, tagsMap, fieldsMap);
					
					
					
					CacheObject cacheObject = cacheChannel.get(Constant.REGION, tableName);
					
					if (cacheObject.getValue() == null) {
						cacheChannel.set(Constant.REGION, tableName, fieldsMap);
					} else {
						HashMap<String, Object> cacheData = (HashMap<String, Object>) cacheObject.getValue();
						cacheData.putAll(fieldsMap);
						cacheChannel.set(Constant.REGION, tableName, cacheData);
					}
					
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
    };
}
