package com.walker.tcp.support;

import com.walker.infrastructure.utils.StringUtils;
import com.walker.tcp.Connection;
import com.walker.tcp.ConnectionCallback;
import com.walker.tcp.ConnectionManager;
import com.walker.tcp.ProtocolResolver;
import com.walker.tcp.Response;
import com.walker.tcp.TcpEngine;
import com.walker.tcp.lb.LongConnectionMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public abstract  class AbstractTcpLongEngine implements TcpEngine {

	protected final transient Logger logger = LoggerFactory.getLogger(getClass());

	private int port = 5800;
	private String name;
	private ConnectionManager connectionManager;

	private boolean openHeartBeat = true;

	private int id = 0;

	private List<ProtocolResolver<?>> protocolResolverList;

	// 2023-08-24 连接空闲回调对象
	protected ConnectionCallback connectionCallback;

	public void setConnectionCallback(ConnectionCallback connectionCallback) {
		this.connectionCallback = connectionCallback;
	}

	public List<ProtocolResolver<?>> getProtocolResolverList() {
		return protocolResolverList;
	}

	public void setProtocolResolverList(List<ProtocolResolver<?>> protocolResolverList) {
		this.protocolResolverList = protocolResolverList;
	}

	public boolean isOpenHeartBeat() {
		return openHeartBeat;
	}

	/**
	 * 设置是否启动服务端心跳检测线程
	 * @param openHeartBeat
	 */
	public void setOpenHeartBeat(boolean openHeartBeat) {
		this.openHeartBeat = openHeartBeat;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public void setName(String name) {
		this.name = name;
	}

	public void setConnectionManager(ConnectionManager connectionManager) {
		this.connectionManager = connectionManager;
	}

	@Override
	public int getPort() {
		return port;
	}

	@Override
	public String getName() {
		return name;
	}

	@Override
	public boolean supportLongConnection() {
		return true;
	}

	@Override
	public ConnectionManager getConnectionManager() {
		return connectionManager;
	}

	@Override
	public void start() throws Exception {
//		if(StringUtils.isEmptyList(this.protocolResolverList)){
//			throw new IllegalArgumentException(ProtocolResolver.ERR_NOFOUND);
//		}
		try{
			this.onStart();
			startTime = System.currentTimeMillis();
			if(openHeartBeat){
				this.startHeartBeatThread();
			}

		} catch(Exception ex){
			logger.error("启动长连接失败：" + this.getName(), ex);
			throw new Exception(ex);
		}
	}

	private void startHeartBeatThread(){
		if(!started){
			executorService.execute(timerRunner);
			logger.info("长连接'" + name + "'启动......");
			started = true;
		} else {
			throw new IllegalStateException("长连接已启动，调用状态错误。name = " + name);
		}
	}

	@Override
	public void shutdown() throws Exception {
		this.onShutdown();
	}

	@Override
	public String sendResponse(Response<?> response){
		if(!this.supportLongConnection()){
			throw new IllegalStateException("改引擎不支持长连接，不能向客户端推送消息");
		}
		return this.onSendResponse(response);
	}

	/**
	 * 向所有客户端发送广播消息
	 * @param response
	 */
	public void sendBroadcast(Response<?> response){
		List<Connection> clientList = this.connectionManager.queryAllConnectionListBy(getId());
		logger.debug("------> 准备发送广播，engineId = {}, 客户端数量：{}", this.getId(), clientList == null? 0: clientList.size());
		if(!StringUtils.isEmptyList(clientList)){
			for(Connection conn : clientList){
				conn.write(response);
			}
		}
	}

	protected abstract void onStart() throws Exception;
	protected abstract void onShutdown() throws Exception;

	/**
	 * 通过引擎下发指令时，回调该函数
	 * @param response
	 */
	protected abstract String onSendResponse(Response<?> response);

//	/**
//	 * 返回服务端定义的心跳响应，由子类实现，因为涉及到消息的分隔符。
//	 * @return
//	 */
//	protected abstract Response<?> getServerHeartBeatResponse();

	//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	// 检测客户端连接状态的子线程
	//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	long startTime = 0;
	boolean started = false; // 是否已经启动运行
	private int currentFailedTime = 0;

	// 内部时钟频率，默认：5分钟
	private long heartBeatTimeInterval = 1000 * 60 * 5;

	public long getHeartBeatTimeInterval() {
		return heartBeatTimeInterval;
	}

	/**
	 * 设置心跳检测频率，默认：5分钟（5 * 60000毫秒）
	 * @param timeInterval
	 */
	public void setHeartBeatTimeInterval(long timeInterval) {
		this.heartBeatTimeInterval = timeInterval;
	}

	private Thread kernelThread = null;

	private ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
		@Override
		public Thread newThread(Runnable r) {
			kernelThread = new Thread(r);
			kernelThread.setDaemon(true);
			return kernelThread;
		}
	});

	private InteralTimerRunner timerRunner = new InteralTimerRunner();

//	private final Response testResponse = this.getServerHeartBeatResponse();

	private class InteralTimerRunner implements Runnable{

		private boolean stop = false;

		public InteralTimerRunner(){}

//		public void setStop() {
//			this.stop = true;
//		}

		@Override
		public void run() {
			while(!stop){

				try{
//					List<Connection> list = connectionManager.queryAllConnectionList();
					List<Connection> list = connectionManager.queryAllConnectionListBy(AbstractTcpLongEngine.this.getId());
					if(!StringUtils.isEmptyList(list)){
						long currentTime = System.currentTimeMillis();
						for(Connection conn : list){
//							if(!conn.isConnected()){
//								connectionManager.removeConnection(conn.getId());
//								logger.info("连接已断开，服务端移除该对象：" + conn.getName());
//							} else {
//								conn.write(emptyResponse);
//							}
							// 定时检测客户端是否物理连接，直接发空数据尝试
							if(logger.isDebugEnabled()){
								logger.debug(AbstractTcpLongEngine.this.getId() + " 发送心跳数据，name：" + conn.getName());
							}
//							conn.write(getServerHeartBeatResponse());
							if(!(conn instanceof LongConnectionMeta)){
								// 2023-09-26 只有物理连接才能推送心跳
								conn.write(conn.getProtocolResolver().getHeartBeatResponse());
							}

							// 2023-08-24，检测连接通道是否空闲（心跳不算）
							// 目前没有实际使用。
							if(connectionCallback != null && (currentTime - conn.getLastTime())/1000 >= 600){
								connectionCallback.onIdle(conn);
							}
						}
					}
				} catch(Exception ex){
					if(currentFailedTime >= Integer.MAX_VALUE){
						currentFailedTime = 0;
					}
					currentFailedTime ++;
					logger.error("连接检测线程调用失败一次 " , ex);
				} finally {
					try {
						if(heartBeatTimeInterval > 0){
							logger.debug("*************** sleep: " + heartBeatTimeInterval);
							TimeUnit.MILLISECONDS.sleep(heartBeatTimeInterval);
						}
					} catch (InterruptedException e) {}
				}
			}
		}
	}

	@Override
	public long getStartTime() {
		return startTime;
	}

	@Override
	public int getId() {
		return id;
	}

	public void setId(int id){
		if(id <= 0 || id >= Integer.MAX_VALUE){
			throw new IllegalArgumentException("engine id must 0~65536");
		}
		if(this.id > 0){
			throw new IllegalArgumentException("engine id already exist, id = " + this.id);
		}
		this.id = id;
	}

	public long getEngineStartDelaySeconds() {
		return engineStartDelaySeconds;
	}

	/**
	 * 设置引擎，延时启动，多少秒，默认：120
	 * @param engineStartDelaySeconds
	 * @date 2023-09-21
	 */
	public void setEngineStartDelaySeconds(long engineStartDelaySeconds) {
		this.engineStartDelaySeconds = engineStartDelaySeconds;
	}

	private long engineStartDelaySeconds = 120;
}