package com.walker.push.rocketmq;

import com.walker.infrastructure.ApplicationRuntimeException;
import com.walker.infrastructure.core.ApplicationBeanInitialized;
import com.walker.queue.AbstractQueueManager;
import com.walker.queue.QueueException;
import com.walker.tcp.ActionCallable;
import com.walker.tcp.ActionCallablePostProcessor;
import com.walker.tcp.Connection;
import com.walker.tcp.ConnectionManager;
import com.walker.tcp.Request;
import com.walker.tcp.Response;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 基于MQ消息队列的队列管理器。
 * <pre>
 *     1) 目前其实没有用到MQ队列，因为当前仅用于聊天系统中，请求数据不需要添加到队列中。
 *     2) 只有业务调用<code>WebSocketEngine</code>推送时，才会走队列处理。
 * </pre>
 * @author 时克英
 * @date 2023-09-26
 */
public class RocketQueueManager extends AbstractQueueManager implements ApplicationBeanInitialized {

    @Override
    public void startup() {

    }

    @Override
    public void push(String key, Object data, Object option) throws QueueException {
        if(data == null){
            throw new QueueException("request is required!");
        }
        if(!Request.class.isAssignableFrom(data.getClass())){
            throw new QueueException("请求对象必须是: " + Request.class.getName());
        }

        Request<?> request = (Request<?>) data;
        this.executor.execute(new AsyncProcessRequestTask(request));
    }
//    @Override
//    public void push(String key, Object data, Object option) throws QueueException {
//        // key = request.getProtocolNum();
//        if(data == null){
//            throw new QueueException("队列数据data必须提供!");
//        }
//        if(!MqRequest.class.isAssignableFrom(data.getClass())){
//            throw new QueueException("请求对象必须是: " + MqRequest.class.getName());
//        }
//
//        MqRequest request = (MqRequest) data;
//
//        try{
//            // 2023-09-19 暂时同步发送，后续根据情况加入异步方式。
//            this.rocketMQEnhanceTemplate.send(request.getTopic(), StringUtils.EMPTY_STRING, request);
//            logger.debug("加入一个数据到(MQ)队列：{}", data);
//        } catch (Exception ex){
//            logger.error("{}", data);
//            logger.error("同步发送MQ消息异常：" + ex.getMessage(), ex);
//        }
//    }

    private class AsyncProcessRequestTask implements Runnable{

        private Request<?> request;

        public AsyncProcessRequestTask(Request<?> request){
            this.request = request;
        }

        @Override
        public void run() {
            ActionCallable action = ActionCallablePostProcessor.getAction(request.getProtocolNum());
            if(action == null){
                throw new ApplicationRuntimeException("action未定义，protocol = " + request.getProtocolNum());
            }

            try {
                if(!request.isRequireResponse()){
                    // 不需要响应，直接处理业务
                    action.action(request);
                    return;
                }

                // 需要响应处理
                // 2023-09-26，其实请求过程不需要MQ队列，仅仅在业务直接调用推送时需要使用。
                Response<?> response = action.action(request);
                // 2018-09-20 时克英修改，响应对象中自动加入用户唯一编号
                response.setName(request.getName());

                Connection conn = connectionManager.getConnectionByName(request.getName());
                if(conn == null || !conn.isConnected()){
                    processFailed(request);
                    return;
                }

                conn.write(response);
                processSuccess(request);

            } catch (Exception ex){
                // 这里也可以加入提醒机制，待续
                processFailed(request);
                logger.error("执行action(负载)错误：" + ex.getMessage(), ex);
            }
        }
    }

    /**
     * 消息处理失败，回调方法，子类可覆盖重写
     * @param request
     */
    protected void processFailed(Request<?> request){
        logger.error("(负载)tcp消息处理失败，可能连接不存在或者已关闭，记录日志：" + request.toString());
    }

    /**
     * 消息处理成功，回调方法，子类可覆盖重写
     * @param request
     */
    protected void processSuccess(Request<?> request){
        logger.debug("(负载)tcp消息处理成功：" + request.getProtocolNum());
    }


    public void setExecutor(ThreadPoolTaskExecutor executor) {
        this.executor = executor;
    }

    private ThreadPoolTaskExecutor executor = null;

    private ConnectionManager connectionManager;

    public void setConnectionManager(ConnectionManager connectionManager) {
        this.connectionManager = connectionManager;
    }
//    public void setRocketMQEnhanceTemplate(RocketMQEnhanceTemplate rocketMQEnhanceTemplate) {
//        this.rocketMQEnhanceTemplate = rocketMQEnhanceTemplate;
//    }
//    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
}