package cn.reghao.tnb.message.app.rabbit.listener;

import cn.reghao.jutil.jdk.serializer.JsonConverter;
import cn.reghao.tnb.message.api.dto.msg.BaseMessage;
import cn.reghao.tnb.message.app.service.MessageConsumer;
import cn.reghao.tnb.message.app.ws.WsConnection;
import cn.reghao.tnb.message.app.ws.msg.ChatPayload;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author reghao
 * @date 2024-11-28 14:37:18
 */
@Slf4j
@Component
public class RabbitListeners {
    private final WsConnection wsConnection;
    private final MessageConsumer messageConsumer;

    public RabbitListeners(WsConnection wsConnection, MessageConsumer messageConsumer) {
        this.wsConnection = wsConnection;
        this.messageConsumer = messageConsumer;
    }

    @RabbitListener(bindings =@QueueBinding(
            // 队列名字可以随便定义
            value = @Queue(autoDelete = "false"),
            // 交换机的名字必须和生产者保持一致, fanout 是以广播模式(发布订阅模式)
            exchange = @Exchange(value = "amq.fanout", type = ExchangeTypes.FANOUT))
    )
    public void chatMessageConsumer(@Payload String msg) {
        ChatPayload chatPayload = JsonConverter.jsonToObject(msg, ChatPayload.class);
        long receiverId = chatPayload.getReceiverId();
        String content = chatPayload.getContent();
        if (wsConnection.getUserSession(receiverId) != null) {
            log.info("send {} -> {}", receiverId, content);
            try {
                wsConnection.sendChatMessage(receiverId+"", chatPayload);
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            log.info("{} 的 session 不在本节点上", receiverId);
        }
    }

    @RabbitListener(bindings =@QueueBinding(
            value = @Queue(value = "tnb.message.account", durable = "true"),
            key = "tnb.message.account",
            exchange = @Exchange(value = "amq.direct")),
            ackMode = "MANUAL"
    )
    @RabbitHandler
    public void consumeMessage(@Payload BaseMessage baseMessage, Channel channel, Message message) {
        messageConsumer.sendMessage(baseMessage);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            log.info("手动确认消息 {}", deliveryTag);
            channel.basicAck(deliveryTag,true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
