package com.ms.biz.paas;

import com.alibaba.fastjson.JSON;
import com.jinritemai.cloud.base.core.service.MqFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * author       hahafeng
 * email        chenzhifeng.777@bytedance.com
 * date         2022/11/24 5:14 PM
 */
@Slf4j
//@Component
public class MqDemoService implements ApplicationListener<ContextRefreshedEvent> {
    private static final String TOPIC = MqFactory.generateProdTopic("miaoshougoods_test");
    private static final String GROUP = "GID_miaoshougoods_test";

    public void testSendMsg() {
        DefaultMQProducer producer;
        try {
            producer = MqFactory.newMqProducer(GROUP);
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }

        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message(TOPIC,
                        "YOUR MESSAGE TAG",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                log.info("testSendMsg send msg success! {}", JSON.toJSONString(sendResult));
            } catch (Exception e) {
                //消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
                log.error("testSendMsg send msg failed!", e);
                return;
            }
        }
    }

    public void testConsumer() {
        DefaultMQPushConsumer consumer = MqFactory.newMqConsumer(GROUP, TOPIC, new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                log.info("testConsumer Receive New Messages: {}", JSON.toJSONString(messages));
                for (MessageExt message: messages) {
                    message.getBody();
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        testConsumer();
        testSendMsg();
    }
}
