package com.ms.biz.quque;

import lombok.extern.slf4j.Slf4j;
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;

@Slf4j
public class Producer {
    private final String group;

    public Producer(String group) {
        this.group = group;
    }

    private DefaultMQProducer getGroupProducer() {
        try {
            return MqFactory.newMqProducer(group);
        } catch (MQClientException e) {
            log.error("New producer error:", e);
            throw new RuntimeException("Can not find producer!");
        }
    }

    public SendResult send(Message msg, long timeout) {
        log.info("Send message keys: " + msg.getKeys());
        log.info("Send message topic: " + msg.getTopic());
        log.info("Send message body: " + new String(msg.getBody()));
        try {
            return getGroupProducer().send(msg, timeout);
        } catch (Throwable e) {
            //消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
            log.error("Send message failed!", e);
            throw new RuntimeException("Send message failed");
        }
    }
}
