package com.iplatform.tcp.lb;

import com.iplatform.base.Constants;
import com.iplatform.base.PlatformRuntimeException;
import com.iplatform.tcp.util.ws.LoginResponse;
import com.iplatform.tcp.util.ws.WebDataResponse;
import com.walker.infrastructure.utils.JsonUtils;
import com.walker.push.rocketmq.tcp.MqResponse;
import com.walker.tcp.Connection;
import com.walker.tcp.ConnectionManager;
import com.walker.tcp.Response;
import com.walker.tcp.lb.LongConnectionMeta;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * MQ消息监听接收处理。
 * <pre>
 *     1) consumerGroup：消费者组，这个是针对同一个主题和Tag。
 *     资料介绍:
 *     A.一个消费者组，代表着一群topic相同，tag相同（即逻辑相同）的Consumer。通过一个消费者组，则可容易的进行负载均衡以及容错
 *     B.使用时，一个节点下，一个topic加一个tag可以对应一个consumer。一个消费者组就是横向上多个节点的相同consumer为一个消费组。
 * </pre>
 * @date 2023-10-07 修改消费者组，尝试解决多个队列消息无法接收问题。
 * @date 2023-12-14 监听对象放在业务中处理，平台注释该对象，业务通过MQ接收其他节点发送的聊天消息。如果要测试，请打开注解。（标注废弃是为了引起注意！）
 */
//@RocketMQMessageListener(topic = "${iplatform.tcp.connection-host}", selectorExpression = "*", consumerGroup = "${iplatform.tcp.connection-host}-consumer-group")
@Deprecated
public class SimpleMqListener implements RocketMQListener<String> {

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

    @Override
    public void onMessage(String s) {
        logger.debug("mq = {}", s);
//        MqResponse mqResponse = null;
        Map<String, Object> mqResponse = null;
        try {
//            mqResponse = JsonUtils.jsonStringToObject(s, MqResponse.class);
            mqResponse = JsonUtils.jsonStringToObject(s, Map.class);
        } catch (Exception e) {
            throw new PlatformRuntimeException("mq消息转换json对象失败:" + e.getMessage(), e);
        }

        MqResponse message = MqResponseUtils.acquireMqResponse(mqResponse);

        // 超过最大重试次数时调用子类方法处理
        if (message.getRetryTimes() > getMaxRetryTimes()) {
            handleMaxRetriesExceeded(message);
            return;
        }

        Response<?> response = null;
        try {
            response = this.translateResponse(mqResponse.get("response").toString());
        } catch (Exception e) {
            throw new PlatformRuntimeException("转换成'Response'失败：" + e.getMessage(), e);
        }

        try {
            long now = System.currentTimeMillis();
//            handleMessage(message);
            Connection conn = connectionManager.getConnectionByName(response.getName());
            if(conn == null || !conn.isConnected()){
                logger.debug("mq消息已接收，但长连接不存在无法推送，response = {}", response);
                return;
            }

            if(conn instanceof LongConnectionMeta){
                throw new IllegalStateException("这个应该是本地物理连接，但找到的是：LongConnectionMeta，name=" + response.getName());
            }
            conn.write(response);

            long costTime = System.currentTimeMillis() - now;
            logger.debug("消息{}消费成功，耗时[{}ms]", message.getKey(), costTime);

        } catch (Exception e) {
            logger.error("消息{}消费异常", message.getKey(),e);
            // 是捕获异常还是抛出，由子类决定
            if (throwException()) {
                //抛出异常，由DefaultMessageListenerConcurrently类处理
                throw new RuntimeException(e);
            }
            //此时如果不开启重试机制，则默认ACK了
//            if (isRetry()) {
//                handleRetry(message);
//            }
        }
    }

    /**
     * 转换成实际发送的业务数据，业务可以继承该类，并重写方法。
     * @param json
     * @return
     * @throws Exception
     */
    protected Response<?> translateResponse(String json) throws Exception{
        Map<String, Object> responseMap = JsonUtils.jsonStringToObject(json, Map.class);
        if(!responseMap.containsKey(KEY_PROTOCOL_NUM)){
            throw new IllegalArgumentException("responseMap中必须包含协议号字段:" + KEY_PROTOCOL_NUM);
        }
        String protocolNum = responseMap.get(KEY_PROTOCOL_NUM).toString();
        if(protocolNum.equals("login")){
            return JsonUtils.jsonStringToObject(json, LoginResponse.class);
        } else if(protocolNum.equals(Constants.PUSH_SCOPE_DATA)){
            return JsonUtils.jsonStringToObject(json, WebDataResponse.class);
        } else {
            throw new UnsupportedOperationException("未实现的 websocket.response对象转换：" + json);
        }
    }

    protected void handleMaxRetriesExceeded(MqResponse message) {

    }

    protected boolean throwException() {
        return true;
    }

    /**
     * 最大重试次数
     *
     * @return 最大重试次数，默认5次
     */
    protected int getMaxRetryTimes() {
        return MAX_RETRY_TIMES;
    }

    /**
     * 默认重试次数
     */
    private static final int MAX_RETRY_TIMES = 3;

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

    private ConnectionManager connectionManager;
    private final String KEY_PROTOCOL_NUM = "protocolNum";
}