package com.ms.biz.quque;

import java.util.Objects;
import java.util.function.Consumer;

import com.ms.biz.tool.SystemTool;
import org.apache.logging.log4j.util.Strings;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MqFactory {
    private static final Logger log = LoggerFactory.getLogger(MqFactory.class);

    private MqFactory() {
    }

    public static DefaultMQProducer newMqProducer(String group) throws MQClientException {
        String key = "mq-producer-" + group;
        Object ins = InstanceHolder.getIns(key);
        if (!Objects.isNull(ins)) {
            return (DefaultMQProducer)ins;
        } else {
            DefaultMQProducer producer = new DefaultMQProducer(group, getAclRPCHook(), true, null);
            producer.setNamesrvAddr(getMqUrl());

            try {
                producer.start();
                log.info("#newMqProducer start success! GID: {}", group);
                InstanceHolder.setIns(key, producer);
                return producer;
            } catch (MQClientException var5) {
                log.error("#newMqProducer start went wrong! GID: " + group, var5);
                throw var5;
            }
        }
    }

    public static DefaultMQPushConsumer newMqConsumer(String group, String topic, MessageListenerConcurrently messageListenerConcurrently) {
        return newMqConsumer(group, topic, messageListenerConcurrently, null);
    }

    public static DefaultMQPushConsumer newMqConsumer(String group, String topic, MessageListenerConcurrently messageListenerConcurrently, Consumer<DefaultMQPushConsumer> configConsumer) {
        String key = "mq-consumer-" + group + topic;
        Object ins = InstanceHolder.getIns(key);
        if (!Objects.isNull(ins)) {
            return (DefaultMQPushConsumer)ins;
        } else {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group, getAclRPCHook(), new AllocateMessageQueueAveragely(), true, (String)null);
            consumer.setNamesrvAddr(getMqUrl());
            if (!Objects.isNull(configConsumer)) {
                configConsumer.accept(consumer);
            }
            consumer.setInstanceName(group + "_" + topic);

            String logInfo = String.format(" GID: %s , topic: %s", group, topic);

            try {
                consumer.subscribe(topic, "*");
            } catch (MQClientException var10) {
                log.error("#newMqConsumer subscribe went wrong !" + logInfo, var10);
                throw new RuntimeException(var10);
            }

            consumer.registerMessageListener(messageListenerConcurrently);

            try {
                consumer.start();
            } catch (MQClientException var9) {
                log.error("#newMqConsumer start went wrong !" + logInfo, var9);
                throw new RuntimeException(var9);
            }

            log.info("#newMqConsumer start success! {}", logInfo);
            InstanceHolder.setIns(key, consumer);
            return consumer;
        }
    }

    public static String generateProdTopic(String topic) {
        return getMqInstanceId() + "%" + topic;
    }

    private static String getMqInstanceId() {
        return SystemTool.isDevEnv() ? "local" : System.getenv("sys-rmq-id");
    }

    private static String getMqUrl() {
        String url = "127.0.0.1";
        String port = "9876";
        if (!SystemTool.isDevEnv()) {
            url = System.getenv("sys-rmq-url");
            port = System.getenv("sys-rmq-port");
        }
        return !Strings.isEmpty(url) && !Strings.isEmpty(port) ? String.format("http://%s:%s", url, port) : null;
    }

    private static RPCHook getAclRPCHook() {
        if (SystemTool.isDevEnv()) {
            return null;
        }
        return new AclClientRPCHook(new SessionCredentials(getAK(), getSK()));
    }

    private static String getAK() {
        return SystemTool.isDevEnv() ? "" : System.getenv("sys-rmq-ak");
    }

    private static String getSK() {
        return SystemTool.isDevEnv() ? "" : System.getenv("sys-rmq-sk");
    }
}
