package com.alibaba.rocketmq.client.consumer.cacheable;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.producer.concurrent.DefaultLocalMessageStore;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicLong;

public class CacheableConsumer
{
    private static final Logger LOGGER = ClientLogger.getLog();

    private String consumerGroupName;

    private DefaultLocalMessageStore localMessageStore;

    private final ConcurrentHashMap<String, MessageHandler> topicHandlerMap;

    private static final AtomicLong CONSUMER_NAME_COUNTER = new AtomicLong();

    private static final String BASE_INSTANCE_NAME = "CacheableConsumer@";

    private DefaultMQPushConsumer defaultMQPushConsumer;

    private boolean started;

    private static final int CORE_POOL_SIZE_FOR_DELAY_TASKS = 2;

    private static final int CORE_POOL_SIZE_FOR_WORK_TASKS = 10;

    private int corePoolSizeForDelayTasks = CORE_POOL_SIZE_FOR_DELAY_TASKS;

    private int corePoolSizeForWorkTasks = CORE_POOL_SIZE_FOR_WORK_TASKS;

    private ScheduledExecutorService scheduledExecutorDelayService = Executors
            .newScheduledThreadPool(corePoolSizeForDelayTasks);

    private ScheduledExecutorService scheduledExecutorWorkerService = Executors
            .newScheduledThreadPool(corePoolSizeForWorkTasks);

    private static String getInstanceName() {
        try {
            return BASE_INSTANCE_NAME + InetAddress.getLocalHost().getHostAddress() + "_" + CONSUMER_NAME_COUNTER.incrementAndGet();
        } catch (UnknownHostException e) {
            return BASE_INSTANCE_NAME + "127.0.0.1_" + CONSUMER_NAME_COUNTER.incrementAndGet();
        }
    }

    public CacheableConsumer(String consumerGroupName) {
        this.consumerGroupName = consumerGroupName;
        this.topicHandlerMap = new ConcurrentHashMap<String, MessageHandler>();
        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroupName);
        localMessageStore = new DefaultLocalMessageStore(consumerGroupName);
        defaultMQPushConsumer.setInstanceName(getInstanceName());
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    }

    public CacheableConsumer registerMessageHandler(MessageHandler messageHandler) throws MQClientException {
        if (started) {
            throw new IllegalStateException("Please register before start");
        }

        if (null == messageHandler.getTopic() || messageHandler.getTopic().trim().isEmpty()) {
            throw new RuntimeException("Topic cannot be null or empty");
        }

        topicHandlerMap.putIfAbsent(messageHandler.getTopic(), messageHandler);
        defaultMQPushConsumer.subscribe(messageHandler.getTopic(),
                null != messageHandler.getTag() ? messageHandler.getTag() : "*");
        return this;
    }

    public CacheableConsumer registerMessageHandler(Collection<MessageHandler> messageHandlers)
            throws MQClientException {
        for (MessageHandler messageHandler : messageHandlers) {
            registerMessageHandler(messageHandler);
        }
        return this;
    }

    public void start() throws InterruptedException, MQClientException {
        if (topicHandlerMap.isEmpty()) {
            throw new RuntimeException("Please at least configure one message handler to subscribe one topic");
        }

        MessageListenerConcurrently frontController = new FrontController(topicHandlerMap,
                scheduledExecutorWorkerService, scheduledExecutorDelayService);
        defaultMQPushConsumer.registerMessageListener(frontController);
        defaultMQPushConsumer.start();
        started = true;
        LOGGER.debug("DefaultMQPushConsumer starts.");
    }

    public void setConsumerGroupName(String consumerGroupName) {
        this.consumerGroupName = consumerGroupName;
    }

    public boolean isStarted() {
        return started;
    }

    public void setCorePoolSizeForDelayTasks(int corePoolSizeForDelayTasks) {
        this.corePoolSizeForDelayTasks = corePoolSizeForDelayTasks;
    }

    public void setCorePoolSizeForWorkTasks(int corePoolSizeForWorkTasks) {
        this.corePoolSizeForWorkTasks = corePoolSizeForWorkTasks;
    }
}

@@ -455,24 +455,92 @@
 eFromWhere;%0A
+import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;%0A
 import org.s
@@ -552,16 +552,16 @@
 Logger;%0A
-
 %0Aimport 
@@ -1352,16 +1352,176 @@
 arted;%0A%0A
+    private MessageModel messageModel = MessageModel.BROADCASTING;%0A%0A    private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;%0A%0A
     priv
@@ -2843,16 +2843,77 @@
 ame());%0A
+        defaultMQPushConsumer.setMessageModel(messageModel);%0A
         
@@ -2954,17 +2954,17 @@
 omWhere(
-C
+c
 onsumeFr
@@ -2974,33 +2974,8 @@
 here
-.CONSUME_FROM_LAST_OFFSET
 );%0A 
@@ -4940,32 +4940,32 @@
 ForWorkTasks) %7B%0A
-
         this.cor
@@ -5019,11 +5019,736 @@
 s;%0A    %7D
+%0A%0A    public void setMessageModel(MessageModel messageModel) %7B%0A        this.messageModel = messageModel;%0A%0A        if (started) %7B%0A            throw new RuntimeException(%22Please set message model before start%22);%0A        %7D%0A%0A        if (null != defaultMQPushConsumer) %7B%0A            defaultMQPushConsumer.setMessageModel(messageModel);%0A        %7D%0A    %7D%0A%0A    public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) %7B%0A        this.consumeFromWhere = consumeFromWhere;%0A%0A        if (started) %7B%0A            throw new RuntimeException(%22Please set consume-from-where before start%22);%0A        %7D%0A%0A        if (null != defaultMQPushConsumer) %7B%0A            defaultMQPushConsumer.setConsumeFromWhere(consumeFromWhere);%0A        %7D%0A    %7D
 %0A%7D%0A

Add configuration for messageModel and consumeFromWhere.